flowmon-export-ipfix.c

Go to the documentation of this file.

static const char rcsid[] __attribute__((used)) = "$Id: flowmon-export-ipfix.c 928 2013-01-16 08:32:47Z 255519 $";

/* This is necessary to enable GNU extension for fprintf: "%as" calls malloc */
#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <endian.h>
#include <errno.h>

#include <flowmonexp/plugin_export.h>
#include "flowmon-export-ipfix.h"

SET_PLUGIN_TYPE(PLUGIN_TYPE_EXPORT);

/*********** USER FUNCTIONS ************/

/* Forward declarations */
static int fei_send_packet(ipfix_packet_t *packet, plugin_private_t *conf);


static void fei_init_configuration(plugin_private_t *conf)
{
        conf->getters = NULL;
        conf->templates = NULL;
        conf->sequenceNum = 0;
        conf->templatesDataSize = 0;
        conf->counter = 0;
        conf->exportedPackets = 0;

        /* Set params */
        conf->host = NULL;
        conf->templateFile = NULL;
        conf->port = NULL;
        conf->protocol = 0;
        conf->ip = AF_UNSPEC;
        conf->flags = 0;
        conf->reconnectTimeout = 0;
        conf->lastReconnect = 0;
        conf->odid = 0;
        conf->templateRefreshTime = 0;
        conf->templateRefreshPackets = 0;
}

static int fei_parse_params(char *params, plugin_private_t *conf)
{
        char *token, *value, *param;
        uint8_t state = 0; /* state 0: expect param, state 1: expect value */

        /* Check that there are parameters */
        if (params == NULL) {
                fprintf(stderr, "IPFIX export plugin need parameters\n");
                return 1;
        }

        /* Parse parameter string */
        token = strtok(params, ",=");
        while (token != NULL) {
                if (state == 0) {
                        param = token;
                } else {
                        value = token;
                        MSG("param: %s\t", param);
                        MSG("value: %s\n", value);

                        if (strcmp("host", param) == 0) {
                                conf->host = value;
                        } else if (strcmp("template-file", param) == 0) {
                                conf->templateFile = value;
                        } else if (strcmp("port", param) == 0) {
                                conf->port = value;
                        } else if (strcmp("protocol", param) == 0) {
                                if (strcmp("UDP", value) == 0) {
                                        conf->protocol = IPPROTO_UDP;
                                } else if (strcmp("TCP", value) == 0) {
                                        conf->protocol = IPPROTO_TCP;
                                } else if (strcmp("SCTP", value) == 0) {
                                        conf->protocol = IPPROTO_SCTP;
                                } else {
                                        fprintf(stderr, "Unknown protocol %s\n", value);
                                        return 1;
                                }
                        } else if (strcmp("ip", param) == 0) {
                                int ip = atoi(value);
                                switch (ip) {
                                case 4: conf->ip = AF_INET; break;
                                case 6: conf->ip = AF_INET6; break;
                                case 64:
                                        conf->ip = AF_INET6;
                                        conf->flags = AI_V4MAPPED | AI_ALL;
                                        break;
                                default: conf->ip = AF_UNSPEC; break;
                                }
                        } else if (strcmp("reconnect-timeout", param) == 0) {
                                conf->reconnectTimeout = atoi(value);
                        } else if (strcmp("odid", param) == 0) {
                                conf->odid = atoi(value);
                        } else if (strcmp("template-refresh-time", param) == 0) {
                                conf->templateRefreshTime = atoi(value);
                        } else if (strcmp("template-refresh-packets", param) == 0) {
                                conf->templateRefreshPackets = atoi(value);
                        } else {
                                fprintf(stderr, "Unknown parameter '%s'\n", param);
                                return 1;
                        }
                }

                /* Move to next token */
                token = strtok(NULL, ",=");
                state = 1 - state;
        }

        /* Check that every parameter has its value */
        if (state == 1) {
                fprintf(stderr, "'%s' does not have a value\n", param);
                return 1;
        }

        /* Set default values */
        if (conf->port == NULL) {
                printf("Using default port 4739\n");
                conf->port = "4739";
        }
        if (conf->protocol == 0) {
                printf("Using default protocol UDP\n");
                conf->protocol = IPPROTO_UDP;
        }
        if (conf->reconnectTimeout < 1) {
                conf->reconnectTimeout = RECONNECT_TIMEOUT;
        }
        if (conf->host == NULL) {
                printf("Using default hostname 'localhost'\n");
                conf->host = "localhost";
        }
        if (conf->protocol == IPPROTO_UDP && conf->templateRefreshTime == 0) {
                printf("Using default template refresh time '600s'\n");
                conf->templateRefreshTime = 600;
        }
        if (conf->templateFile == NULL) {
                printf("Using default template file "DEFAULT_TEMPLATE_FILE"\n");
                conf->templateFile = DEFAULT_TEMPLATE_FILE;
        }

        return 0;
}

static template_file_record_t *fei_get_template_record_by_name(const char *name, plugin_private_t *conf)
{
        template_file_record_t *tmpFileRecord = conf->templateFileRecords;

        if (name == NULL) {
                MSG("Cannot get template for NULL name\n");
                return NULL;
        }

        while (tmpFileRecord && tmpFileRecord->getterName) {
                if (strcmp(name, tmpFileRecord->getterName) == 0) {
                        return tmpFileRecord;
                }
                tmpFileRecord++;
        }

        return NULL;
}

static void fei_store_getters(flow_record_getter_t *getter_list, plugin_private_t *conf)
{
        flow_record_getter_t *tmp = getter_list;

        conf->getters = NULL;

        while (tmp && tmp->name) {
                MSG("getter: %s - %i", tmp->name, tmp->length);
                /* Find appropriate template file record */
                template_file_record_t *tmpFileRecord = fei_get_template_record_by_name(tmp->name, conf);
                /* Use only allowed getters */
                if (tmpFileRecord != NULL) {
                        MSG(" (used)");
                        getter_copy_to(&conf->getters, tmp);
                }
                MSG("\n");
                tmp++;
        }
}

static int fei_load_template_file(plugin_private_t *conf)
{
        int res, line = 1, countLines = 0, c;
        template_file_record_t *tfr = NULL;
        FILE *f;

        /* Open the template file */
        f = fopen(conf->templateFile, "r");
        if (!f) {
                fprintf(stderr, "Cannot open template file %s\n", conf->templateFile);
                return 1;
        }

        /* Count the number of lines in the file */
        do {
                c = fgetc(f);
                if (c == '\n') countLines++;
        } while (c != EOF);
        fseek(f, SEEK_SET, 0);

        /* Allocate big enough buffer */
        conf->templateFileRecords = calloc((countLines + 1), sizeof(template_file_record_t));
        if (!conf->templateFileRecords) {
                return 1;
        }
        tfr = conf->templateFileRecords;

        while (1) {
                /* Fill the structure */
                /* In newer versions of glibc, 'm' modificator is defined instead of 'a' for dynamically allocated strings */
                res = fscanf(f, "%as %hu %hu %d\n", &tfr->getterName, &tfr->enterpriseNumber, &tfr->elementID, &tfr->length);

                /* Check the result */
                if (res == EOF) {
                        /* End of file */
                        if (tfr->getterName) free(tfr->getterName);
                        tfr->getterName = NULL;
                        tfr = NULL;
                        break;
                } else if (res < 3 || tfr->getterName[0] == '#') {
                        /* Not enough parameters or comment, skip line */
                        if (tfr->getterName) free(tfr->getterName);
                        tfr->getterName = NULL;
                        line++;
                        continue;
                }

                MSG("Template element: %s, %i, %i, %i\n", tfr->getterName, tfr->enterpriseNumber, tfr->elementID, tfr->length);

                line++;
                tfr++;
        }


        return 0;
}

static void fei_init_template_buffer(template_t *tmpl)
{
        *((uint16_t *) &tmpl->buffer[0]) = htons(tmpl->id);
        /* Length will be updated later */
        /* *((uint16_t *) &tmpl->buffer[2]) = htons(0); */
        tmpl->bufferSize = 4;
}

static template_t *fei_create_template(flow_record_t *record, plugin_private_t *conf)
{
        flow_record_getter_t *tmpGetter = conf->getters;
        uint16_t maxID = FIRST_TEMPLATE_ID, len;
        template_t *tmpTemplate = conf->templates, *newTemplate;

        /* Create new template structure */
        newTemplate = (template_t *) malloc(sizeof(template_t));
        if (!newTemplate) {
                return NULL;
        }
        newTemplate->templateGetters = NULL;
        newTemplate->fieldCount = 0;
        newTemplate->recordCount = 0;

        /* Set template ID to maximum + 1 */
        while (tmpTemplate != NULL) {
                if (tmpTemplate->id >= maxID) maxID = tmpTemplate->id + 1;
                tmpTemplate = tmpTemplate->next;
        }
        newTemplate->id = maxID;
        ((uint16_t *) newTemplate->templateRecord)[0] = htons(newTemplate->id);

        /* Template header size */
        newTemplate->templateSize = 4;

        /* For every valid getter */
        while (tmpGetter && tmpGetter->name) {
                if (tmpGetter->valid(tmpGetter->self, record)) {

                        /* Find appropriate template file record */
                        template_file_record_t *tmpFileRecord = fei_get_template_record_by_name(tmpGetter->name, conf);
                        if (tmpFileRecord != NULL) {
                                /* Set information element ID */
                                uint16_t eID = tmpFileRecord->elementID;
                                if (tmpFileRecord->enterpriseNumber != 0) {
                                        eID |= 0x8000;
                                }
                                *((uint16_t *) &newTemplate->templateRecord[newTemplate->templateSize]) = htons(eID);

                                /* Set element length */
                                if (tmpFileRecord->length == 0) {
                                        len = tmpGetter->length;
                                } else {
                                        len = tmpFileRecord->length;
                                }
                                *((uint16_t *) &newTemplate->templateRecord[newTemplate->templateSize+2]) = htons(len);

                                /* Update template size */
                                newTemplate->templateSize += 4;

                                /* Add enterprise number if required */
                                if (tmpFileRecord->enterpriseNumber != 0) {
                                        *((uint32_t *) &newTemplate->templateRecord[newTemplate->templateSize]) =
                                                htonl(tmpFileRecord->enterpriseNumber);
                                        newTemplate->templateSize += 4;
                                }

                                /* Store the template */
                                getter_copy_to(&newTemplate->templateGetters, tmpGetter);

                                /* Increase field count */
                                newTemplate->fieldCount++;
                        } else {
                                MSG("Cannot find template file record for getter %s\n", tmpGetter->name);
                        }
                }

                tmpGetter++;
        }

        /* Store getter mask */
        newTemplate->mask = getter_bitmap_from_record(conf->getters, record);

        /* Set field count */
        ((uint16_t *) newTemplate->templateRecord)[1] = htons(newTemplate->fieldCount);

        /* Allocate memory for relevant template records loaded from file */
        newTemplate->templateFileRecords = malloc(sizeof(template_file_record_t *)*(newTemplate->fieldCount+1));
        if (!newTemplate->templateFileRecords) {
                free(newTemplate->templateGetters);
                free(newTemplate);
                return NULL;
        }

        /* Store relevant template records loaded from file */
        template_file_record_t **tmpFileRecord = newTemplate->templateFileRecords;
        tmpGetter = newTemplate->templateGetters;
        while (tmpGetter && tmpGetter->name) {
                *tmpFileRecord = fei_get_template_record_by_name(tmpGetter->name, conf);
                tmpGetter++;
                tmpFileRecord++;
        }

        /* Initialize buffer for records */
        fei_init_template_buffer(newTemplate);

        /* Update total template size */
        conf->templatesDataSize += newTemplate->bufferSize;

        /* The template was not exported yet */
        newTemplate->exported = 0;
        newTemplate->exportTime = time(NULL);
        newTemplate->exportPacket = conf->exportedPackets;

        /* Add the new template to the list */
        newTemplate->next = conf->templates;
        conf->templates = newTemplate;

        return newTemplate;
}

static template_t *fei_find_template(flow_record_t *record, plugin_private_t *conf)
{
        uint64_t mask = getter_bitmap_from_record(conf->getters, record);
        template_t *tmpTemplate = conf->templates;

        /* Go over all templates */
        while (tmpTemplate != NULL) {
                MSG("Comparing templates: %lu %lu\n", mask, tmpTemplate->mask);
                /* Return the one with matching mask */
                if (tmpTemplate->mask == mask) {
                        return tmpTemplate;
                }
                tmpTemplate = tmpTemplate->next;
        }

        /* Return NULL when there is no matching template */
        return NULL;
}


static int fei_add_record(flow_record_t *record, template_t *tmpl, plugin_private_t *conf)
{
        flow_record_getter_t *tmpGetter = tmpl->templateGetters;
        template_file_record_t **tmpFileRecord = tmpl->templateFileRecords;
        int32_t elementLen;
        uint16_t recordLen = 0;
        uint8_t variableLength = 0;

        while (tmpGetter && tmpGetter->name) {
                if (tmpGetter->valid(tmpGetter->self, record)) {

                        /* Get element length. From template file or from getter. */
                        if ((*tmpFileRecord)->length == 0) {
                                elementLen = tmpGetter->length;
                        } else {
                                elementLen = (*tmpFileRecord)->length;
                        }

                        /* Handle variable length elements */
                        if (elementLen == -1) {
                                elementLen = tmpGetter->current_length(tmpGetter->self, record);
                                variableLength = 1;
                        } else {
                                variableLength = 0;
                        }

                        /* Check that the buffer is not full yet, variable length can add at most 3 bytes */
                        if (tmpl->bufferSize + recordLen + elementLen + variableLength*3 > TEMPLATE_BUFFER_SIZE) {
                                MSG("Template %i is full\n", tmpl->id);
                                return 1;
                        }

                        /* Trim elements that do not fit the buffer */
                        if (elementLen > PACKET_DATA_SIZE) {
                                fprintf(stderr, "Element %s trimmed to %i bytes\n", tmpGetter->name, PACKET_DATA_SIZE);
                                elementLen = PACKET_DATA_SIZE;
                        }

                        if (variableLength) {
                                /* Handle writing of variable length elements */
                                if (elementLen < 255) {
                                        /*
                                          0                   1                   2                   3
                                          0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                         | Length (< 255)|          Information Element                  |
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                         |                      ... continuing as needed                 |
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                        */

                                        tmpl->buffer[tmpl->bufferSize + recordLen] = elementLen;

                                        /* record contains one byte of length information */
                                        recordLen += 1;
                                } else {
                                        /*
                                          0                   1                   2                   3
                                          0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                         |      255      |      Length (0 to 65535)      |       IE      |
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                         |                      ... continuing as needed                 |
                                         +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                                        */

                                        tmpl->buffer[tmpl->bufferSize + recordLen] = 255;
                                        *((uint16_t *) &tmpl->buffer[tmpl->bufferSize + recordLen + 1]) = htons(elementLen);

                                        /* record contains three bytes of length information */
                                        recordLen += 3;
                                }
                        }

                        /* Fill the value to the buffer, let getter handle the byte ordering */
                        tmpGetter->filler(tmpGetter->self, record,
                                        &tmpl->buffer[tmpl->bufferSize + recordLen], elementLen, 1);

                        recordLen += elementLen;
                } else {
                        MSG("Record added to invalid template\n");
                }

                tmpGetter++;
                tmpFileRecord++;
        }

        /* Check that the packet size does not exceed the limit */
        if (conf->templatesDataSize + recordLen + IPFIX_HEADER_SIZE > PACKET_DATA_SIZE) {
                MSG("Buffered data exceeded maximum packet data size, need to export data\n");
                return 1;
        }

        conf->templatesDataSize += recordLen;
        tmpl->bufferSize += recordLen;
        tmpl->recordCount++;

        return 0;
}

static int fei_fill_ipfix_header(char *ptr, uint16_t size, plugin_private_t *conf)
{
        ipfix_header_t *header = (ipfix_header_t *)ptr;

        header->version = htons(IPFIX_VERISON);
        header->length = htons(size);
        header->exportTime = htonl(time(NULL));
        header->sequenceNumber = htonl(conf->sequenceNum);
        header->observationDomainId = htonl(conf->odid);

        return IPFIX_HEADER_SIZE;
}

static int fei_fill_template_set_header(char *ptr, uint16_t size)
{

        ipfix_template_set_header_t *header = (ipfix_template_set_header_t *) ptr;

        header->id = htons(TEMPLATE_SET_ID);
        header->length = htons(size);

        return IPFIX_SET_HEADER_SIZE;
}

static void fei_check_template_lifetime(template_t *tmpl, plugin_private_t *conf)
{
        if (conf->templateRefreshTime != 0 &&
                        conf->templateRefreshTime + tmpl->exportTime <= time(NULL)) {
                MSG("Template %i refresh time expired (%is)\n", tmpl->id, conf->templateRefreshTime);
                tmpl->exported = 0;
        }

        if (conf->templateRefreshPackets != 0 &&
                        conf->templateRefreshPackets + tmpl->exportPacket <= conf->exportedPackets) {
                MSG("Template %i refresh packets expired (%i packets)\n", tmpl->id, conf->templateRefreshPackets);
                tmpl->exported = 0;
        }
}

static uint16_t fei_create_template_packet(ipfix_packet_t *packet, plugin_private_t *conf)
{
        template_t *tmp = conf->templates;
        uint16_t totalSize = 0;
        char *ptr;

        /* Get total size */
        while (tmp != NULL) {
                /* Check UDP template lifetime */
                if (conf->protocol == IPPROTO_UDP) {
                        fei_check_template_lifetime(tmp, conf);
                }
                if (tmp->exported == 0) {
                        totalSize += tmp->templateSize;
                }
                tmp = tmp->next;
        }

        /* Check that there are templates to export */
        if (totalSize == 0) {
                return 0;
        }

        totalSize += IPFIX_HEADER_SIZE+IPFIX_SET_HEADER_SIZE;

        /* Allocate memory for the packet */
        packet->data = malloc(sizeof(char)*(totalSize));
        if (!packet->data) {
                return 0;
        }
        ptr = packet->data;

        /* Create ipfix message header */
        ptr += fei_fill_ipfix_header(ptr, totalSize, conf);
        /* Create template set header */
        ptr += fei_fill_template_set_header(ptr, totalSize-IPFIX_HEADER_SIZE);


        /* Copy the templates to the packet */
        tmp = conf->templates;
        while (tmp != NULL) {
                if (tmp->exported == 0) {
                        memcpy(ptr, tmp->templateRecord, tmp->templateSize);
                        ptr += tmp->templateSize;
                        /* Set the templates as exported, store time and serial number */
                        tmp->exported = 1;
                        tmp->exportTime = time(NULL);
                        tmp->exportPacket = conf->exportedPackets;
                }
                tmp = tmp->next;
        }

        packet->length = totalSize;
        packet->flows = 0;

        return totalSize;
}


static uint16_t fei_create_data_packet(ipfix_packet_t *packet, plugin_private_t *conf)
{
        template_t *tmp = conf->templates;
        uint16_t totalSize = IPFIX_HEADER_SIZE; /* Include IPFIX header to total size */
        uint32_t deltaSequenceNum = 0; /* Number of exported records in this packet */
        static char buff[PACKET_DATA_SIZE];
        char *ptr;

        /* Pointer to internal buffer is returned */
        packet->data = buff;

        /* Start adding data after the header */
        ptr = buff + totalSize;

        /* Copy the data sets to the packet */
        conf->templatesDataSize = 0; /* Erase total data size */
        while (tmp != NULL) {
                /* Add only templates with data that fits to one packet */
                if (tmp->recordCount > 0 && totalSize + tmp->bufferSize <= PACKET_DATA_SIZE) {
                        memcpy(ptr, tmp->buffer, tmp->bufferSize);
                        /* Set SET length */
                        ((ipfix_template_set_header_t *) ptr)->length = htons(tmp->bufferSize);
                        MSG("Adding template %i of length %i to data packet\n", tmp->id, tmp->bufferSize);
                        ptr += tmp->bufferSize;
                        /* Count size of the data copied to buffer */
                        totalSize += tmp->bufferSize;
                        /* Delete data from buffer */
                        tmp->bufferSize = IPFIX_SET_HEADER_SIZE;

                        /* Store number of exported records  */
                        deltaSequenceNum += tmp->recordCount;
                        tmp->recordCount = 0;
                }
                /* Update total data size, include empty template buffers (only set headers) */
                conf->templatesDataSize += tmp->bufferSize;
                tmp = tmp->next;
        }

        /* Check that there are packets to export */
        if (totalSize == IPFIX_HEADER_SIZE) {
                return 0;
        }

        /* Create ipfix message header at the beginning */
        fei_fill_ipfix_header(buff, totalSize, conf);

        /* Fill number of flows and size of the packet */
        packet->flows = deltaSequenceNum;
        packet->length = totalSize;

        return totalSize;
}

static void fei_expire_templates(plugin_private_t *conf)
{
        template_t *tmp;
        for (tmp = conf->templates; tmp != NULL; tmp = tmp->next) {
                tmp->exported = 0;
                if (conf->protocol == IPPROTO_UDP) {
                        tmp->exportTime = time(NULL);
                        tmp->exportPacket = conf->exportedPackets;
                }
        }
}

static void fei_send_templates(plugin_private_t *conf)
{
        ipfix_packet_t pkt;

        /* Send all new templates */
        if (fei_create_template_packet(&pkt, conf)) {
                /* Send template packet */
                /* After error, the plugin sends all templates after reconnection,
                 * so we need not concern about it here */
                fei_send_packet(&pkt, conf);

                free(pkt.data);
        }
}

static void fei_send_data(plugin_private_t *conf)
{
        ipfix_packet_t pkt;

        /* Send all new templates */
        if (fei_create_data_packet(&pkt, conf)) {
                if (fei_send_packet(&pkt, conf) == 1) {
                        /* Collector reconnected, resend the packet */
                        fei_send_packet(&pkt, conf);
                }
        }
}

static int fei_connect(plugin_private_t *conf)
{
        struct addrinfo hints, *tmp;

        memset(&hints, 0, sizeof(hints));
        hints.ai_family = conf->ip;
        hints.ai_protocol = conf->protocol;
        hints.ai_flags = AI_ADDRCONFIG | conf->flags;

        /* get server address */
        if (getaddrinfo(conf->host, conf->port, &hints, &conf->addrinfo) != 0) {
                perror("Cannot get server info");
                return 1;
        }

        /* Try addrinfo strucutres one by one */
        for (tmp = conf->addrinfo; tmp != NULL; tmp = tmp->ai_next) {

                if (tmp->ai_family != AF_INET && tmp->ai_family != AF_INET6) {
                        continue;
                }

                /* Print information about target address */
                char buff[INET6_ADDRSTRLEN];
                inet_ntop(tmp->ai_family,
                                (tmp->ai_family == AF_INET) ?
                                                (void *) &((struct sockaddr_in *) tmp->ai_addr)->sin_addr :
                                                (void *) &((struct sockaddr_in6 *) tmp->ai_addr)->sin6_addr,
                                (char *) &buff, sizeof(buff));

                printf("Connecting to IP %s\n", buff);

                MSG("Socket configuration: AI Family: %i, AI Socktype: %i, AI Protocol: %i\n",
                        tmp->ai_family, tmp->ai_socktype, tmp->ai_protocol);

                /* create socket */
                conf->socket = socket(conf->addrinfo->ai_family,
                                conf->addrinfo->ai_socktype, conf->addrinfo->ai_protocol);
                if (conf->socket == -1) {
                        perror("Cannot create new socket");
                        continue;
                }

                /* connect to server with TCP and SCTP */
                if (conf->protocol != IPPROTO_UDP &&
                                connect(conf->socket, conf->addrinfo->ai_addr, conf->addrinfo->ai_addrlen) == -1) {
                        perror("Cannot connect to collector");
                        close(conf->socket);
                        continue;
                }

                /* Connected, meaningless for UDP */
                if (conf->protocol != IPPROTO_UDP) {
                        printf("Successfully connected to collector\n");
                }
                break;
        }

        /* Return error when all addrinfo structures were tried*/
        if (tmp == NULL) {
                /* Free allocated resources */
                freeaddrinfo(conf->addrinfo);
                return 1;
        }

        return 0;
}

static int fei_reconnect(plugin_private_t *conf)
{
        /* Check for broken connection */
        if (conf->lastReconnect != 0) {
                /* Check whether we need to attempt reconnection */
                if (conf->lastReconnect + conf->reconnectTimeout <= time(NULL)) {
                        /* Try to reconnect */
                        if (fei_connect(conf) == 0) {
                                conf->lastReconnect = 0;
                                /* Resend all templates */
                                fei_expire_templates(conf);
                                fei_send_templates(conf);
                        } else {
                                /* Set new reconnect time and drop packet */
                                conf->lastReconnect = time(NULL);
                                return 1;
                        }
                } else {
                        /* Timeout not reached, drop packet */
                        return 1;
                }
        }

        return 0;
}

static int fei_send_packet(ipfix_packet_t *packet, plugin_private_t *conf)
{
        int ret; /* Return value of sendto */
        int sent = 0; /* Sent data size */

        /* Check that connection is OK or drop packet */
        if (fei_reconnect(conf)) {
                return -1;
        }

        /* sendto() does not guarantee that everything will be send in one piece */
        while (sent < packet->length) {
                /* Send data to collector (TCP and SCTP ignores last two arguments) */
                ret = sendto(conf->socket, (void *) packet->data + sent, packet->length - sent, 0,
                                conf->addrinfo->ai_addr, conf->addrinfo->ai_addrlen);

                /* Check that the data were sent correctly */
                if (ret == -1) {
                        switch (errno) {
                        case 0: break; /* OK */
                        case ECONNRESET:
                        case EINTR:
                        case ENOTCONN:
                        case ENOTSOCK:
                        case EPIPE:
                        case EHOSTUNREACH:
                        case ENETDOWN:
                        case ENETUNREACH:
                        case ENOBUFS:
                        case ENOMEM:

                                /* The connection is broken */
                                fprintf(stderr, "Collector closed connection\n");

                                /* free resources */
                                close(conf->socket);
                                freeaddrinfo(conf->addrinfo);

                                /* Set last connection try time so that we would reconnect immediatelly */
                                conf->lastReconnect = 1;

                                /* Reset the sequences number since it is unique per connection */
                                conf->sequenceNum = 0;
                                ((ipfix_header_t *) packet->data)->sequenceNumber = 0; /* no need to change byteorder of 0 */

                                /* Say that we should try to connect and send data again */
                                return 1;
                        default:
                                /* Unknown error */
                                perror("Cannot send data to collector");
                                return -1;
                        }
                }

                /* No error from sendto(), add sent data count to total */
                sent += ret;
        }

        /* Update sequence number for next packet */
        conf->sequenceNum += packet->flows;

        /* Increase packet counter */
        conf->exportedPackets++;

        MSG("Packet (%li) sent to %s on port %s. Next sequence number is %i\n",
                conf->exportedPackets, conf->host, conf->port, conf->sequenceNum);

        return 0;
}


/*********** API FUNCTIONS ************/

plugin_desc_t *plugin_export_desc()
{
        return &plugin_desc;
}


void *plugin_export_init(char *params, flow_record_getter_t *getter_list)
{
        plugin_private_t *conf;

        MSG("IPFIX export plugin init start\n");

        /* Allocate private configuration structure */
        conf = malloc(sizeof(plugin_private_t));
        if (!conf) {
                return NULL;
        }

        /* Init plugin configuration */
        fei_init_configuration(conf);

        /* Parse input parameters */
        if (fei_parse_params(params, conf)) {
                return NULL; /* TODO How to end correctly? */
        }

        /* Connect to collector */
        if (fei_connect(conf) != 0) {
                return NULL;
        }

        /* Load template file */
        if (fei_load_template_file(conf)) {
                return NULL;
        }

        /* Store getters that will be used */
        fei_store_getters(getter_list, conf);

        MSG("IPFIX export plugin init end\n");

        return conf;
}


int plugin_export_export(void *plugin_private, flow_record_t *record)
{
        plugin_private_t *conf = (plugin_private_t *) plugin_private;

#ifdef DEBUG
        flow_record_getter_t *tmp = conf->getters;
        unsigned char block[1024];
        int i;

        MSG("FLOW: %li, ", conf->counter++);
        while (tmp && tmp->name) {
                if (tmp->valid(tmp->self, record)) {
                        int len = tmp->current_length(tmp->self, record);
                        if (len > 1024) {
                                len = 1024;
                        }
                        MSG("%s: ", tmp->name);
                        tmp->filler(tmp->self, record, block, len, 0);
                                for (i = 0; i < len; i++) {
                                        MSG("%02X", block[i]);
                                }
                        MSG(", ");
                }
                tmp++;
        }
        MSG("\n");
#endif

        /* Get template for the record */
        template_t *tmpl = fei_find_template(record, conf);
        if (!tmpl) {
                MSG("Creating new template\n");
                tmpl = fei_create_template(record, conf);
                MSG("Template created\n");
                if (!tmpl) {
                        MSG("Cannot create new template\n");
                }
        }

        MSG("Adding record to template\n");
        if (fei_add_record(record, tmpl, conf) == 1) {
                /* Flush data when record is too big to fit */

                /* Send all new templates */
                fei_send_templates(conf);

                /* Send the data packet */
                fei_send_data(conf);

                /* Try to add the record again */
                fei_add_record(record, tmpl, conf);
        }

        return 0;
}

int plugin_export_flush(void *plugin_private)
{
        plugin_private_t *conf = (plugin_private_t *) plugin_private;

        MSG("plugin flush\n");
        fflush(stdout);

        /* Send all new templates */
        fei_send_templates(conf);

        /* Send the data packet */
        fei_send_data(conf);

        return 0;
}

/* TODO Callback for close function is missing.
 * It is not possible to withdraw templates, free addrinfo and close connection properly. */