sFlow Mailing List: sflowtool 3.9 contrib: MySQL Support

From: Horsington, Colin <colin.horsington@transact.com.au>
Date: 10/19/05
Message-ID: <A80938E906BC3340B0E7C4731F4B355E010A5E83@semailbox02.actewagl.production.com.au>

Hi All,

After looking at various sFlow to MySQL packages, I thought to integrate
basic MySQL support into the standard sflowtools package. This is done
at configuration time with the flag "--with-mysql". It's release one
and could be polished by more experienced C coders :).

Below are the diffs for the following files:
        o configure
        o src/sflowtool.c
        o src/sflowtool.h
A new files:
        o src/sql.h
        o src/sflow.sql (table definition)

... hopefully these will make it into the standard sflowtool package. I
also add a new argument "-x" which includes the AS information in an
eXtended line output format and is basically the same information as the
SQL inserts.

If there are any questions please let me know.

Cheers,

Colin Horsington

------------------------------------------------------------------------
-------------------------
------------------------------------------------------------------------
-------------------------

@@ -1268,7 +1271,7 @@

PACKAGE=sflowtool
-VERSION=3.9
+VERSION=3.9.1
am__api_version="1.6"
ac_aux_dir=
for ac_dir in $srcdir $srcdir/.. $srcdir/../..; do
@@ -4238,7 +4241,12 @@
}'
fi

-DEFS=-DHAVE_CONFIG_H
+if test -n "$WITH_MYSQL"; then
+ DEFS="-DHAVE_CONFIG_H -DHAVE_LIBMYSQLCLIENT=1 -DWITH_MYSQL=1";
+ LIBS="-lmysqlclient";
+else
+ DEFS="-DHAVE_CONFIG_H";
+fi

ac_libobjs=
ac_ltlibobjs=

------------------------------------------------------------------------
-------------------------
------------------------------------------------------------------------
-------------------------

--- sflowtool-3.9/src/sflowtool.c 2005-09-20 04:08:45.000000000
+1000
+++ sflowtool-3.9.1/src/sflowtool.c 2005-10-19 18:43:48.000000000
+1000
@@ -12,6 +12,11 @@
#include <time.h>
#include <setjmp.h>

+#ifdef WITH_MYSQL
+#include <mysql/mysql.h>
+#include "sql.h"
+#endif
+
#ifdef WIN32
#include "winsock2.h"
#include "fcntl.h"
@@ -115,6 +120,17 @@
u_int32_t linktype; /* data link type (DLT_*) */
};

+#ifdef WITH_MYSQL
+/* mysql db information */
+typedef struct _db_settings {
+ char *username;
+ char *password;
+ char *host;
+ char *database;
+ char *type; // "MySQL" for the moment
+} db_settings;
+#endif
+
struct pcap_pkthdr {
   struct timeval ts; /* time stamp */
   u_int32_t caplen; /* length of portion present */
@@ -136,7 +152,7 @@
   int sock;
} SFForwardingTarget;

-typedef enum { SFLFMT_FULL=0, SFLFMT_PCAP, SFLFMT_LINE } EnumSFLFormat;
+typedef enum { SFLFMT_FULL=0, SFLFMT_PCAP, SFLFMT_LINE,
SFLFMT_LINE_EXTENDED } EnumSFLFormat;

typedef struct _SFConfig {
   /* sflow options */
@@ -156,6 +172,10 @@
   u_int32_t tcpdumpHdrPad;
   u_char zeroPad[100];
   int pcapSwap;
+#ifdef WITH_MYSQL
+ /* db settings */
+ struct _db_settings db_settings;
+#endif

#ifdef SPOOFSOURCE
int spoofSource;
@@ -483,6 +503,125 @@
}

/*_________________---------------------------__________________
+ _________________ writeFlowLineExtended __________________
+ -----------------___________________________------------------
+*/
+
+static void writeFlowLineExtended(SFSample *sample)
+{
+
+ char agentIP[51], srcIP[51], dstIP[51];
+ // source
+ printf("FLOW,%s,%d,%d,",
+ printAddress(&sample->agent_addr, agentIP, 50),
+ sample->inputPort,
+ sample->outputPort);
+ // layer 2
+
printf("%02x%02x%02x%02x%02x%02x,%02x%02x%02x%02x%02x%02x,0x%04x,%d,%d",
+ sample->eth_src[0],
+ sample->eth_src[1],
+ sample->eth_src[2],
+ sample->eth_src[3],
+ sample->eth_src[4],
+ sample->eth_src[5],
+ sample->eth_dst[0],
+ sample->eth_dst[1],
+ sample->eth_dst[2],
+ sample->eth_dst[3],
+ sample->eth_dst[4],
+ sample->eth_dst[5],
+ sample->eth_type,
+ sample->in_vlan,
+ sample->out_vlan);
+ // layer 3/4
+ printf(",%s,%s,%d,0x%02x,%d,%d,%d,0x%02x",
+ IP_to_a(sample->dcd_srcIP.s_addr, srcIP),
+ IP_to_a(sample->dcd_dstIP.s_addr, dstIP),
+ sample->dcd_ipProtocol,
+ sample->dcd_ipTos,
+ sample->dcd_ipTTL,
+ sample->dcd_sport,
+ sample->dcd_dport,
+ sample->dcd_tcpFlags);
+ // bytes
+ printf(",%d,%d,%d",
+ sample->sampledPacketSize,
+ sample->sampledPacketSize - sample->stripped -
sample->offsetToIPV4,
+ sample->meanSkipCount);
+ // Extended Information
+ // Colin
+ //printf(",%d,",sample->my_as);
+
+ printf(",%d,%d,%d,%d,%d,%d,%d,%d\n",
+ sample->my_as,
+ sample->srcMask,
+ sample->dstMask,
+ sample->src_as,
+ sample->src_peer_as,
+ sample->dst_as_path_len,
+ sample->dst_peer_as,
+ sample->dst_as);
+}
+
+
+#ifdef WITH_MYSQL
+/*_________________---------------------------__________________
+ _________________ writeMySQLRecord __________________
+ -----------------___________________________------------------
+*/
+
+static void writeMySQLRecord(SFSample *sample)
+{
+ char agentIP[51], srcIP[51], dstIP[51];
+ /* do mysql insert */
+ char buf[2048];
+ sprintf(buf, sflow_to_mysql,
+ printAddress(&sample->agent_addr, agentIP, 50),
+ sample->inputPort,
+ sample->outputPort,
+ sample->eth_src[0],
+ sample->eth_src[1],
+ sample->eth_src[2],
+ sample->eth_src[3],
+ sample->eth_src[4],
+ sample->eth_src[5],
+ sample->eth_dst[0],
+ sample->eth_dst[1],
+ sample->eth_dst[2],
+ sample->eth_dst[3],
+ sample->eth_dst[4],
+ sample->eth_dst[5],
+ sample->eth_type,
+ sample->in_vlan,
+ sample->out_vlan,
+ IP_to_a(sample->dcd_srcIP.s_addr, srcIP),
+ IP_to_a(sample->dcd_dstIP.s_addr, dstIP),
+ sample->dcd_ipProtocol,
+ sample->dcd_ipTos,
+ sample->dcd_ipTTL,
+ sample->dcd_sport,
+ sample->dcd_dport,
+ sample->dcd_tcpFlags,
+ sample->sampledPacketSize,
+ sample->sampledPacketSize - sample->stripped -
sample->offsetToIPV4,
+ sample->meanSkipCount,
+ sample->my_as,
+ sample->srcMask,
+ sample->dstMask,
+ sample->src_as,
+ sample->src_peer_as,
+ sample->dst_as_path_len,
+ sample->dst_peer_as,
+ sample->dst_as);
+
+ if(mysql_real_query(&mysql, buf, strlen(buf))){
+ fprintf(stderr, "mysql_real_query() failed\n");
+ fprintf(stderr, "%s\n", mysql_error(&mysql));
+ }
+}
+#endif
+
+/*_________________---------------------------__________________
   _________________ writeCountersLine __________________
   -----------------___________________________------------------
*/
@@ -1386,6 +1525,9 @@
     }
     sf_log("\n");
   }
+ // Colin
+ //printf("my_as: %d\n",sample->my_as);
+
}

/*_________________---------------------------__________________
@@ -1879,7 +2021,7 @@
     sample->ds_index = samplerId & 0x00ffffff;
     sf_log("sourceId %lu:%lu\n", sample->ds_class, sample->ds_index);
   }
-
+
   sample->meanSkipCount = getData32(sample);
   sample->samplePool = getData32(sample);
   sample->dropEvents = getData32(sample);
@@ -1917,7 +2059,9 @@
       case INMEXTENDED_SWITCH: readExtendedSwitch(sample); break;
       case INMEXTENDED_ROUTER: readExtendedRouter(sample); break;
       case INMEXTENDED_GATEWAY:
- if(sample->datagramVersion == 2) readExtendedGateway_v2(sample);
+ if(sample->datagramVersion == 2) {
+ readExtendedGateway_v2(sample);
+ }
        else readExtendedGateway(sample);
        break;
       case INMEXTENDED_USER: readExtendedUser(sample); break;
@@ -1926,13 +2070,19 @@
       }
     }
   }
-
+
   /* if we are exporting netflow and we have an IPv4 layer, compose the
datagram now */
   if(sfConfig.netFlowOutputSocket && sample->gotIPV4)
sendNetFlowDatagram(sample);
   /* if we are writing tcpdump format, write the next packet record now
*/
   if(sfConfig.outputFormat == SFLFMT_PCAP) writePcapPacket(sample);
+ /* or line-by-line extended output... */
+ if(sfConfig.outputFormat == SFLFMT_LINE_EXTENDED)
writeFlowLineExtended(sample);
   /* or line-by-line output... */
   if(sfConfig.outputFormat == SFLFMT_LINE) writeFlowLine(sample);
+#ifdef WITH_MYSQL
+ /* write record to mysql if compiled in */
+ if(sfConfig.db_settings.type == "MySQL") writeMySQLRecord(sample);
+#endif
}

/*_________________---------------------------__________________
@@ -2034,6 +2184,12 @@
   if(sfConfig.outputFormat == SFLFMT_PCAP) writePcapPacket(sample);
   /* or line-by-line output... */
   if(sfConfig.outputFormat == SFLFMT_LINE) writeFlowLine(sample);
+ /* or line-by-line extended output... */
+ if(sfConfig.outputFormat == SFLFMT_LINE_EXTENDED)
writeFlowLineExtended(sample);
+#ifdef WITH_MYSQL
+ /* write record to mysql if compiled in */
+ if(sfConfig.db_settings.type == "MySQL") writeMySQLRecord(sample);
+#endif
}

/*_________________---------------------------__________________
@@ -2572,6 +2728,38 @@
return YES;
}

+
+#ifdef WITH_MYSQL
+/*_________________---------------------------__________________
+ _________________ db connect __________________
+ -----------------___________________________------------------
+*/
+int db_connect()
+{
+ printf("Yay!\n");
+ printf("%s\n",sfConfig.db_settings.host);
+ int result;
+ int store_data;
+ store_data = 1;
+
+ if(store_data == 1){
+ /*
+ * Connect MySQL
+ */
+ mysql_init(&mysql);
+ mysql_options(&mysql,MYSQL_READ_DEFAULT_GROUP,"sflow");
+ result = (int)mysql_real_connect(&mysql,
sfConfig.db_settings.host, sfConfig.db_settings.username,
+ sfConfig.db_settings.password,
sfConfig.db_settings.database, 0, NULL, 0);
+ if (result == (int)NULL){
+ fprintf(stderr, "Failed to connect to database:
Error: %s\n", mysql_error(&mysql));
+ exit(1);
+ }
+ }
+
+ return 0;
+}
+#endif
+
/*_________________---------------------------__________________
   _________________ instructions __________________
   -----------------___________________________------------------
@@ -2589,6 +2777,7 @@
   fprintf(stderr,"\n");
   fprintf(stderr,"csv output:\n");
   fprintf(stderr, " -l - (output in line-by-line
format)\n"); fprintf(stderr,"\n");
+ fprintf(stderr, " -x - (output in extended
line-by-line format)\n"); fprintf(stderr,"\n");
   fprintf(stderr,"tcpdump output:\n");
   fprintf(stderr, " -t - (output in binary
tcpdump(1) format)\n");
   fprintf(stderr, " -r file - (read binary tcpdump(1)
format)\n");
@@ -2603,6 +2792,12 @@
#ifdef SPOOFSOURCE
   fprintf(stderr, " -S - spoof source of netflow
packets to input agent IP\n");
#endif
+#ifdef WITH_MYSQL
+ fprintf(stderr, " -H host - MySQL server host IP
address\n");
+ fprintf(stderr, " -D database - Database name (table name
is assumed to be 'sflow')\n");
+ fprintf(stderr, " -U username - Database Username\n");
+ fprintf(stderr, " -P password - Database Password\n");
+#endif
   exit(1);
}

@@ -2626,6 +2821,7 @@
     case 'p': sfConfig.sFlowInputPort = atoi(argv[arg++]); break;
     case 't': sfConfig.outputFormat = SFLFMT_PCAP; break;
     case 'l': sfConfig.outputFormat = SFLFMT_LINE; break;
+ case 'x': sfConfig.outputFormat = SFLFMT_LINE_EXTENDED; break;
     case 'r': sfConfig.readPcapFileName = strdup(argv[arg++]); break;
     case 'z': sfConfig.tcpdumpHdrPad = atoi(argv[arg++]); break;
     case 'c':
@@ -2641,6 +2837,13 @@
     case 'd': sfConfig.netFlowOutputPort = atoi(argv[arg++]); break;
     case 'e': sfConfig.netFlowPeerAS = YES; break;
     case 's': sfConfig.disableNetFlowScale = YES; break;
+#ifdef WITH_MYSQL
+ /* db_settings part */
+ case 'H': sfConfig.db_settings.host = strdup(argv[arg++]); break;
+ case 'D': sfConfig.db_settings.database = strdup(argv[arg++]);
break;
+ case 'U': sfConfig.db_settings.username = strdup(argv[arg++]);
break;
+ case 'P': sfConfig.db_settings.password = strdup(argv[arg++]);
break;
+#endif
#ifdef SPOOFSOURCE
     case 'S': sfConfig.spoofSource = YES; break;
#endif
@@ -2650,6 +2853,19 @@
     default: instructions(*argv);
     }
   }
+#ifdef WITH_MYSQL
+ if(sfConfig.db_settings.host && sfConfig.db_settings.username &&
sfConfig.db_settings.database){
+ /* Assume "" as default password */
+ sfConfig.db_settings.password = "";
+ db_connect();
+ sfConfig.db_settings.type = "MySQL";
+ }
+ else {
+ fprintf(stderr, "ERROR: Database settings incorrect.\n");
+ instructions(*argv);
+ }
+
+#endif
}

/*_________________---------------------------__________________

------------------------------------------------------------------------
-------------------------
------------------------------------------------------------------------
-------------------------

--- sflowtool-3.9/src/sflowtool.h 2005-09-20 03:51:25.000000000
+1000
+++ sflowtool-3.9.1/src/sflowtool.h 2005-10-19 18:32:03.000000000
+1000
@@ -58,6 +58,10 @@
INMAddress_value address;
} INMAddress;

+#ifdef WITH_MYSQL
+MYSQL mysql;
+#endif
+
/* Packet header data */

#define INM_MAX_HEADER_SIZE 256 /* The maximum sampled header size.
*/

------------------------------------------------------------------------
-------------------------
------------------------------------------------------------------------
-------------------------
cat sflowtool-3.9.1/src/sql.h
/*
   sFlowtool modification to insert into mysql database
   Author: Colin and Nathan, TransACT Communications
   Date: 17-10-2005

#define sflow_to_mysql \
"INSERT INTO sflow (\
agent_ip,\
in_int,\
out_int,\
src_mac,\
dst_mac,\
eth_type,\
in_vlan,\
out_vlan,\
src_ip,\
dst_ip,\
ip_protocol,\
ip_tos,\
ip_ttl,\
src_port,\
dst_port,\
tcp_flags,\
packet_size,\
ip_size,\
sample_rate,\
src_mask,\
dst_mask,\
my_as,\
src_as,\
src_peer_as,\
dst_as_path_len,\
dst_peer_as,\
dst_as) \
VALUES\
(\
'%s',\
%d,\
%d,\
'%02x%02x%02x%02x%02x%02x',\
'%02x%02x%02x%02x%02x%02x',\
'0x%04x',\
%d,\
%d,\
'%s',\
'%s',\
%d,\
'0x%02x',\
%d,\
%d,\
%d,\
'0x%02x',\
%d,\
%d,\
%d,\
%d,\
%d,\
%d,\
%d,\
%d,\
%d,\
%d,\
%d)"

------------------------------------------------------------------------
-------------------------
------------------------------------------------------------------------
-------------------------
cat sflowtool-3.9.1/src/sflow.sql
DROP TABLE IF EXISTS sflow;
CREATE TABLE sflow (
id int(10) unsigned NOT NULL auto_increment,
PRIMARY KEY (id),

        agent_ip varchar(64),
        in_int int(4),
        out_int int(4),
        src_mac varchar(16),
        dst_mac varchar(16),
        eth_type varchar(16),
        in_vlan int(4),
        out_vlan int(4),
        src_ip varchar(16),
        dst_ip varchar(16),
        ip_protocol int(3),
        ip_tos varchar(16),
        ip_ttl int(3),
        src_port int(5),
        dst_port int(5),
        tcp_flags varchar(16),
        packet_size int(5),
        ip_size int(5),
        sample_rate int(6),
        src_mask int(2),
        dst_mask int(2),
        my_as int(5),
        src_as int(5),
        src_peer_as int(5),
        dst_as_path_len int(2),
        dst_peer_as int(5),
        dst_as int(5),

key
(agent_ip,in_int,out_int,in_vlan,out_vlan,src_ip,dst_ip,ip_protocol,src_
port,dst_port,my_as,src_as,src_peer_as,dst_peer_as,dst_as)
);

************************************************************************
*PLEASE NOTE* This email and any attachments may
be confidential. If received in error, please delete all
copies and advise the sender. The reproduction or
dissemination of this email or its attachments is
prohibited without the consent of the sender.

WARNING RE VIRUSES: Our computer systems sweep
outgoing email to guard against viruses, but no warranty
is given that this email or its attachments are virus free.
Before opening or using attachments, please check for
viruses. Our liability is limited to the re-supply of any
affected attachments.

Any views expressed in this message are those of the
individual sender, except where the sender expressly,
and with authority, states them to be the views of the
organisation.
************************************************************************
Received on Wed Oct 19 02:01:27 2005