static inline ipfix_field_specifier_t *
flowprobe_template_common_fields (ipfix_field_specifier_t * f)
{
-#define flowprobe_template_common_field_count() 3
+#define flowprobe_template_common_field_count() 5
/* ingressInterface, TLV type 10, u32 */
f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
ingressInterface, 4);
packetDeltaCount, 8);
f++;
+ /* flowStartNanoseconds, TLV type 156, u64 */
+ f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
+ flowStartNanoseconds, 8);
+ f++;
+
+ /* flowEndNanoseconds, TLV type 157, u64 */
+ f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
+ flowEndNanoseconds, 8);
+ f++;
+
return f;
}
static inline ipfix_field_specifier_t *
flowprobe_template_l4_fields (ipfix_field_specifier_t * f)
{
-#define flowprobe_template_l4_field_count() 2
+#define flowprobe_template_l4_field_count() 3
/* sourceTransportPort, TLV type 7, u16 */
f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
sourceTransportPort, 2);
f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
destinationTransportPort, 2);
f++;
+ /* tcpControlBits, TLV type 6, u16 */
+ f->e_id_length = ipfix_e_id_length (0 /* enterprise */ ,
+ tcpControlBits, 2);
+ f++;
+
return f;
}
STATIC_ASSERT (sizeof (flowprobe_key_t) == FLOWPROBE_KEY_IN_U32 *
sizeof (u32), "flowprobe_key_t padding is wrong");
+typedef struct
+{
+ u32 sec;
+ u32 nsec;
+} timestamp_nsec_t;
+
typedef struct
{
flowprobe_key_t key;
u64 packetcount;
u64 octetcount;
+ timestamp_nsec_t flow_start;
+ timestamp_nsec_t flow_end;
f64 last_updated;
f64 last_exported;
u32 passive_timer_handle;
+ union
+ {
+ struct
+ {
+ u16 flags;
+ } tcp;
+ } prot;
} flowprobe_entry_t;
/**
#include <vppinfra/error.h>
#include <flowprobe/flowprobe.h>
#include <vnet/ip/ip6_packet.h>
+#include <vlibmemory/api.h>
static void flowprobe_export_entry (vlib_main_t * vm, flowprobe_entry_t * e);
return which;
}
+/*
+ * NTP rfc868 : 2 208 988 800 corresponds to 00:00 1 Jan 1970 GMT
+ */
+#define NTP_TIMESTAMP 2208988800L
+
static inline u32
flowprobe_common_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
{
clib_memcpy (to_b->data + offset, &packetdelta, sizeof (u64));
offset += sizeof (u64);
+ /* flowStartNanoseconds */
+ u32 t = clib_host_to_net_u32 (e->flow_start.sec + NTP_TIMESTAMP);
+ clib_memcpy (to_b->data + offset, &t, sizeof (u32));
+ offset += sizeof (u32);
+ t = clib_host_to_net_u32 (e->flow_start.nsec);
+ clib_memcpy (to_b->data + offset, &t, sizeof (u32));
+ offset += sizeof (u32);
+
+ /* flowEndNanoseconds */
+ t = clib_host_to_net_u32 (e->flow_end.sec + NTP_TIMESTAMP);
+ clib_memcpy (to_b->data + offset, &t, sizeof (u32));
+ offset += sizeof (u32);
+ t = clib_host_to_net_u32 (e->flow_end.nsec);
+ clib_memcpy (to_b->data + offset, &t, sizeof (u32));
+ offset += sizeof (u32);
+
return offset - start;
}
clib_memcpy (to_b->data + offset, &e->key.dst_port, 2);
offset += 2;
+ /* tcp control bits */
+ u16 control_bits = htons (e->prot.tcp.flags);
+ clib_memcpy (to_b->data + offset, &control_bits, 2);
+ offset += 2;
+
return offset - start;
}
static inline void
add_to_flow_record_state (vlib_main_t * vm, vlib_node_runtime_t * node,
flowprobe_main_t * fm, vlib_buffer_t * b,
- u64 timestamp, u16 length,
+ timestamp_nsec_t timestamp, u16 length,
flowprobe_variant_t which, flowprobe_trace_t * t)
{
if (fm->disabled)
ip4_header_t *ip4 = 0;
ip6_header_t *ip6 = 0;
udp_header_t *udp = 0;
+ tcp_header_t *tcp = 0;
+ u8 tcp_flags = 0;
if (flags & FLOW_RECORD_L3 || flags & FLOW_RECORD_L4)
{
if (collect_ip6 && ethertype == ETHERNET_TYPE_IP6)
{
ip6 = (ip6_header_t *) (eth + 1);
- udp = (udp_header_t *) (ip6 + 1);
if (flags & FLOW_RECORD_L3)
{
k.src_address.as_u64[0] = ip6->src_address.as_u64[0];
k.dst_address.as_u64[1] = ip6->dst_address.as_u64[1];
}
k.protocol = ip6->protocol;
+ if (k.protocol == IP_PROTOCOL_UDP)
+ udp = (udp_header_t *) (ip6 + 1);
+ else if (k.protocol == IP_PROTOCOL_TCP)
+ tcp = (tcp_header_t *) (ip6 + 1);
+
octets = clib_net_to_host_u16 (ip6->payload_length)
+ sizeof (ip6_header_t);
}
if (collect_ip4 && ethertype == ETHERNET_TYPE_IP4)
{
ip4 = (ip4_header_t *) (eth + 1);
- udp = (udp_header_t *) (ip4 + 1);
if (flags & FLOW_RECORD_L3)
{
k.src_address.ip4.as_u32 = ip4->src_address.as_u32;
k.dst_address.ip4.as_u32 = ip4->dst_address.as_u32;
}
k.protocol = ip4->protocol;
+ if ((flags & FLOW_RECORD_L4) && k.protocol == IP_PROTOCOL_UDP)
+ udp = (udp_header_t *) (ip4 + 1);
+ else if ((flags & FLOW_RECORD_L4) && k.protocol == IP_PROTOCOL_TCP)
+ tcp = (tcp_header_t *) (ip4 + 1);
+
octets = clib_net_to_host_u16 (ip4->length);
}
- if ((flags & FLOW_RECORD_L4) && udp &&
- (k.protocol == IP_PROTOCOL_TCP || k.protocol == IP_PROTOCOL_UDP))
+
+ if (udp)
{
k.src_port = udp->src_port;
k.dst_port = udp->dst_port;
}
+ else if (tcp)
+ {
+ k.src_port = tcp->src_port;
+ k.dst_port = tcp->dst_port;
+ tcp_flags = tcp->flags;
+ }
if (t)
{
{
e = flowprobe_create (my_cpu_number, &k, &poolindex);
e->last_exported = now;
+ e->flow_start = timestamp;
}
}
else
e->packetcount++;
e->octetcount += octets;
e->last_updated = now;
-
+ e->flow_end = timestamp;
+ e->prot.tcp.flags |= tcp_flags;
if (fm->active_timer == 0
|| (now > e->last_exported + fm->active_timer))
flowprobe_export_entry (vm, e);
u32 n_left_from, *from, *to_next;
flowprobe_next_t next_index;
flowprobe_main_t *fm = &flowprobe_main;
- u64 now;
+ timestamp_nsec_t timestamp;
- now = (u64) ((vlib_time_now (vm) - fm->vlib_time_0) * 1e9);
- now += fm->nanosecond_time_0;
+ unix_time_now_nsec_fraction (×tamp.sec, ×tamp.nsec);
from = vlib_frame_vector_args (frame);
n_left_from = frame->n_vectors;
u16 ethertype0 = clib_net_to_host_u16 (eh0->type);
if (PREDICT_TRUE ((b0->flags & VLIB_BUFFER_FLOW_REPORT) == 0))
- add_to_flow_record_state (vm, node, fm, b0, now, len0,
+ add_to_flow_record_state (vm, node, fm, b0, timestamp, len0,
flowprobe_get_variant
(which, fm->context[which].flags,
ethertype0), 0);
u16 ethertype1 = clib_net_to_host_u16 (eh1->type);
if (PREDICT_TRUE ((b1->flags & VLIB_BUFFER_FLOW_REPORT) == 0))
- add_to_flow_record_state (vm, node, fm, b1, now, len1,
+ add_to_flow_record_state (vm, node, fm, b1, timestamp, len1,
flowprobe_get_variant
(which, fm->context[which].flags,
ethertype1), 0);
&& (b0->flags & VLIB_BUFFER_IS_TRACED)))
t = vlib_add_trace (vm, node, b0, sizeof (*t));
- add_to_flow_record_state (vm, node, fm, b0, now, len0,
+ add_to_flow_record_state (vm, node, fm, b0, timestamp, len0,
flowprobe_get_variant
(which, fm->context[which].flags,
ethertype0), t);
return 1e9 * ts.tv_sec + ts.tv_nsec;
}
+always_inline void
+unix_time_now_nsec_fraction (u32 * sec, u32 * nsec)
+{
+ struct timespec ts;
+ syscall (SYS_clock_gettime, CLOCK_REALTIME, &ts);
+ *sec = ts.tv_sec;
+ *nsec = ts.tv_nsec;
+}
+
always_inline f64
unix_usage_now (void)
{
return 0;
}
+always_inline void
+unix_time_now_nsec_fraction (u32 * sec, u32 * nsec)
+{
+}
+
always_inline f64
unix_usage_now (void)
{
import socket
import unittest
import time
+import re
from scapy.packet import Raw
from scapy.layers.l2 import Ether
-from scapy.layers.inet import IP, UDP
+from scapy.layers.inet import IP, TCP, UDP
from scapy.layers.inet6 import IPv6
from framework import VppTestCase, VppTestRunner, running_extended_tests
from vpp_pg_interface import CaptureTimeoutError
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
+from vpp_ip_route import VppIpRoute, VppRoutePath
class VppCFLOW(VppObject):
super(MethodHolder, cls).setUpClass()
try:
# Create pg interfaces
- cls.create_pg_interfaces(range(7))
+ cls.create_pg_interfaces(range(9))
# Packet sizes
cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
cls.pg3.resolve_arp()
cls.pg4.config_ip4()
cls.pg4.resolve_arp()
+ cls.pg7.config_ip4()
+ cls.pg8.config_ip4()
+ cls.pg8.configure_ipv4_neighbors()
cls.pg5.config_ip6()
cls.pg5.resolve_ndp()
p /= IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
else:
p /= IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
- p /= (UDP(sport=1234, dport=4321) / Raw(payload))
+ p /= UDP(sport=1234, dport=4321)
+ p /= Raw(payload)
info.data = p.copy()
self.extend_packet(p, pkt_size)
self.pkts.append(p)
return p
-class Timers(MethodHolder):
+class Flowprobe(MethodHolder):
"""Template verification, timer tests"""
def test_0001(self):
""" timer less than template timeout"""
- self.logger.info("FFP_TEST_START_0002")
+ self.logger.info("FFP_TEST_START_0001")
self.pg_enable_capture(self.pg_interfaces)
self.pkts = []
# template packet should arrive immediately
templates = ipfix.verify_templates(ipfix_decoder)
- self.create_stream(packets=2)
+ self.create_stream(packets=1)
self.send_packets()
- capture = self.pg2.get_capture(2)
+ capture = self.pg2.get_capture(1)
# make sure the one packet we expect actually showed up
cflow = self.wait_for_cflow_packet(self.collector, templates[1], 15)
self.verify_cflow_data(ipfix_decoder, capture, cflow)
ipfix.remove_vpp_config()
- self.logger.info("FFP_TEST_FINISH_0002")
+ self.logger.info("FFP_TEST_FINISH_0001")
def test_0002(self):
""" timer greater than template timeout"""
- self.logger.info("FFP_TEST_START_0003")
+ self.logger.info("FFP_TEST_START_0002")
self.pg_enable_capture(self.pg_interfaces)
self.pkts = []
self.verify_cflow_data(ipfix_decoder, capture, cflow)
ipfix.remove_vpp_config()
- self.logger.info("FFP_TEST_FINISH_0003")
+ self.logger.info("FFP_TEST_FINISH_0002")
+
+ def test_cflow_packet(self):
+ """verify cflow packet fields"""
+ self.logger.info("FFP_TEST_START_0000")
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pkts = []
+
+ ipfix = VppCFLOW(test=self, intf='pg8', datapath="ip4",
+ layer='l2 l3 l4', active=2)
+ ipfix.add_vpp_config()
+
+ route_9001 = VppIpRoute(self, "9.0.0.0", 24,
+ [VppRoutePath(self.pg8._remote_hosts[0].ip4,
+ self.pg8.sw_if_index)])
+ route_9001.add_vpp_config()
+
+ ipfix_decoder = IPFIXDecoder()
+ templates = ipfix.verify_templates(ipfix_decoder, count=1)
+
+ self.pkts = [(Ether(dst=self.pg7.local_mac,
+ src=self.pg7.remote_mac) /
+ IP(src=self.pg7.remote_ip4, dst="9.0.0.100") /
+ TCP(sport=1234, dport=4321, flags=80) /
+ Raw('\xa5' * 100))]
+
+ nowUTC = int(time.time())
+ nowUNIX = nowUTC+2208988800
+ self.send_packets(src_if=self.pg7, dst_if=self.pg8)
+
+ cflow = self.wait_for_cflow_packet(self.collector, templates[0], 10)
+ self.collector.get_capture(2)
+
+ if cflow[0].haslayer(IPFIX):
+ self.assertEqual(cflow[IPFIX].version, 10)
+ self.assertEqual(cflow[IPFIX].observationDomainID, 1)
+ self.assertEqual(cflow[IPFIX].sequenceNumber, 0)
+ self.assertAlmostEqual(cflow[IPFIX].exportTime, nowUTC, delta=5)
+ if cflow.haslayer(Data):
+ record = ipfix_decoder.decode_data_set(cflow[0].getlayer(Set))[0]
+ # ingress interface
+ self.assertEqual(int(record[10].encode('hex'), 16), 8)
+ # egress interface
+ self.assertEqual(int(record[14].encode('hex'), 16), 9)
+ # packets
+ self.assertEqual(int(record[2].encode('hex'), 16), 1)
+ # src mac
+ self.assertEqual(':'.join(re.findall('..', record[56].encode(
+ 'hex'))), self.pg8.local_mac)
+ # dst mac
+ self.assertEqual(':'.join(re.findall('..', record[80].encode(
+ 'hex'))), self.pg8.remote_mac)
+ flowTimestamp = int(record[156].encode('hex'), 16) >> 32
+ # flow start timestamp
+ self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
+ flowTimestamp = int(record[157].encode('hex'), 16) >> 32
+ # flow end timestamp
+ self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
+ # ethernet type
+ self.assertEqual(int(record[256].encode('hex'), 16), 8)
+ # src ip
+ self.assertEqual('.'.join(re.findall('..', record[8].encode(
+ 'hex'))),
+ '.'.join('{:02x}'.format(int(n)) for n in
+ self.pg7.remote_ip4.split('.')))
+ # dst ip
+ self.assertEqual('.'.join(re.findall('..', record[12].encode(
+ 'hex'))),
+ '.'.join('{:02x}'.format(int(n)) for n in
+ "9.0.0.100".split('.')))
+ # protocol (TCP)
+ self.assertEqual(int(record[4].encode('hex'), 16), 6)
+ # src port
+ self.assertEqual(int(record[7].encode('hex'), 16), 1234)
+ # dst port
+ self.assertEqual(int(record[11].encode('hex'), 16), 4321)
+ # tcp flags
+ self.assertEqual(int(record[6].encode('hex'), 16), 80)
+
+ ipfix.remove_vpp_config()
+ self.logger.info("FFP_TEST_FINISH_0000")
class Datapath(MethodHolder):
Code Review / vpp.git / commitdiff