Feature/ioam6 timestamp

cmd/fluvia/main.go

@@ -51,5 +51,10 @@ func main() {
 		ingressIfName = c.Ipfix.IngressInterface
 	}
 
-	client.New(ingressIfName, raddr)
+	interval := c.Ipfix.Interval
+	if interval <= 0 {
+		interval = 1
+	}
+
+	client.New(ingressIfName, raddr, interval)
 }

go.mod

@@ -5,11 +5,9 @@ go 1.20
 require (
 	github.com/cilium/ebpf v0.11.0
 	github.com/google/gopacket v1.1.19
-	github.com/pkg/errors v0.9.1
+	golang.org/x/sync v0.0.0-20190423024810-112230192c58
+	golang.org/x/sys v0.10.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
-require (
-	golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect
-	golang.org/x/sys v0.10.0 // indirect
-)
+require golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect

go.sum

@@ -6,8 +6,6 @@ github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF
 github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
-github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -17,6 +15,7 @@ golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPI
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

internal/config/config.go

@@ -15,6 +15,7 @@ type Ipfix struct {
 	Address          string `yaml:"address"`
 	Port             string `yaml:"port"`
 	IngressInterface string `yaml:"ingress-interface"`
+	Interval         int    `yaml:"interval"`
 }
 
 type Config struct {

pkg/bpf/bpf.go

@@ -7,45 +7,78 @@
 package bpf
 
 import (
-	"fmt"
+	"errors"
 	"net"
 
 	"github.com/cilium/ebpf"
-	"github.com/pkg/errors"
+	"github.com/cilium/ebpf/link"
+	"github.com/cilium/ebpf/perf"
 )
 
 //go:generate go run github.com/cilium/ebpf/cmd/bpf2go -no-global-types -cc $BPF_CLANG -cflags $BPF_CFLAGS xdp ../../src/main.c -- -I../../src
 
-type XdpProbeData struct {
-	H_dest       [6]uint8
-	H_source     [6]uint8
-	H_proto      uint16
-	_            [2]byte
-	V6Srcaddr    struct{ In6U struct{ U6Addr8 [16]uint8 } }
-	V6Dstaddr    struct{ In6U struct{ U6Addr8 [16]uint8 } }
-	NextHdr      uint8
-	HdrExtLen    uint8
-	RoutingType  uint8
-	SegmentsLeft uint8
-	LastEntry    uint8
-	Flags        uint8
-	Tag          uint16
-	Segments     [10]struct{ In6U struct{ U6Addr8 [16]uint8 } }
+type XdpMetaData struct {
+	ReceivedNano uint64
+	SentSec      uint32
+	SentSubsec   uint32
 }
 
-func ReadXdpObjects(ops *ebpf.CollectionOptions) (*xdpObjects, error) {
+type Xdp struct {
+	objs *xdpObjects
+	link link.Link
+}
+
+func ReadXdpObjects(ops *ebpf.CollectionOptions) (*Xdp, error) {
 	obj := &xdpObjects{}
 	err := loadXdpObjects(obj, ops)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	// TODO: BPF log level remove hardcoding. yaml in config
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
+	}
+
+	return &Xdp{
+		objs: obj,
+	}, nil
+}
+
+func (x *Xdp) Attach(iface *net.Interface) error {
+	l, err := link.AttachXDP(link.XDPOptions{
+		Program:   x.objs.XdpProg,
+		Interface: iface.Index,
+		Flags:     link.XDPGenericMode,
+	})
+	if err != nil {
+		return err
 	}
 
-	return obj, nil
+	x.link = l
+
+	return nil
+}
+
+func (x *Xdp) NewPerfReader() (*perf.Reader, error) {
+	return perf.NewReader(x.objs.PacketProbePerf, 4096)
+}
+
+func (x *Xdp) Close() error {
+	errs := []error{}
+	if err := x.objs.Close(); err != nil {
+		errs = append(errs, err)
+	}
+
+	if err := x.link.Close(); err != nil {
+		errs = append(errs, err)
+	}
+
+	if len(errs) > 0 {
+		return errors.Join(errs...)
+	}
+
+	return nil
 }
 
 const (
@@ -55,16 +88,3 @@ const (
 	XDP_TX
 	XDP_REDIRECT
 )
-
-func PrintEntrys(entry XdpProbeData, count uint64) {
-	mac := func(mac [6]uint8) string {
-		return fmt.Sprintf("%02x:%02x:%02x:%02x:%02x:%02x", mac[0], mac[1], mac[2], mac[3], mac[4], mac[5])
-	}
-	saddr := net.IP(entry.V6Srcaddr.In6U.U6Addr8[:]).String()
-	daddr := net.IP(entry.V6Dstaddr.In6U.U6Addr8[:]).String()
-
-	fmt.Printf(
-		"H_dest: %s, H_source: %v, H_proto: %v, V6Dstaddr: %v, V6Srcaddr: %v -> count: %v\n",
-		mac(entry.H_dest), mac(entry.H_source), entry.H_proto, daddr, saddr, count)
-
-}

pkg/bpf/srv6.go

@@ -3,7 +3,7 @@ package bpf
 import (
 	"encoding/binary"
 	"errors"
-	"net"
+	"net/netip"
 
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
@@ -18,7 +18,7 @@ type Srv6Layer struct {
 	LastEntry    uint8
 	Flags        uint8
 	Tag          uint16
-	Segments     []net.IP
+	Segments     []netip.Addr
 }
 
 var Srv6LayerType = gopacket.RegisterLayerType(
@@ -50,10 +50,11 @@ func (i *Srv6Layer) DecodeFromBytes(data []byte, df gopacket.DecodeFeedback) err
 		startBit := 8 + 16*j
 		endBit := 24 + 16*j
 		var addr []byte
-		for k := endBit; k >= startBit; k-- {
+		for k := startBit; k < endBit; k++ {
 			addr = append(addr, data[k])
 		}
-		i.Segments = append(i.Segments, addr)
+		seg, _ := netip.AddrFromSlice(addr[:16])
+		i.Segments = append(i.Segments, seg)
 	}
 	i.BaseLayer = layers.BaseLayer{
 		Contents: data[:8],
@@ -76,7 +77,8 @@ func (i *Srv6Layer) SerializeTo(b gopacket.SerializeBuffer, opts gopacket.Serial
 	bytes[5] = i.Flags
 	binary.BigEndian.PutUint16(bytes[6:], i.Tag)
 
-	for i2, address := range i.Segments {
+	for i2, seg := range i.Segments {
+		address := seg.AsSlice()
 		lsb := binary.BigEndian.Uint64(address[:8])
 		msb := binary.BigEndian.Uint64(address[8:])
 		binary.BigEndian.PutUint64(bytes[8+16*i2:], lsb)

pkg/client/client.go

@@ -7,11 +7,12 @@ package client
 
 import (
 	"net"
+	"time"
 
-	"github.com/nttcom/fluvia/pkg/packet/ipfix"
+	"github.com/nttcom/fluvia/pkg/ipfix"
 )
 
-func New(ingressIfName string, raddr *net.UDPAddr) ClientError {
+func New(ingressIfName string, raddr *net.UDPAddr, interval int) ClientError {
 	ch := make(chan []ipfix.FieldValue)
 	errChan := make(chan ClientError)
 
@@ -25,7 +26,18 @@ func New(ingressIfName string, raddr *net.UDPAddr) ClientError {
 			}
 		}
 	}()
-	go NewMeter(ingressIfName, ch)
+
+	m := NewMeter(ingressIfName)
+	go func() {
+		err := m.Run(ch, time.Duration(interval))
+		if err != nil {
+			errChan <- ClientError{
+				Component: "meter",
+				Error:     err,
+			}
+		}
+		m.Close()
+	}()
 
 	for {
 		clientError := <-errChan

pkg/client/exporter.go

@@ -10,7 +10,7 @@ import (
 	"net"
 	"os"
 
-	"github.com/nttcom/fluvia/pkg/packet/ipfix"
+	"github.com/nttcom/fluvia/pkg/ipfix"
 )
 
 const OBSERVATION_ID uint32 = 61166
@@ -36,7 +36,7 @@ func (e *Exporter) Run(raddr *net.UDPAddr, flowChan chan []ipfix.FieldValue) err
 	defer conn.Close()
 
 	var m *ipfix.Message
-	// get flow data from go channel
+
 	for {
 		fvs := <-flowChan
 		var sets []ipfix.Set