Merge pull request #153 from HyperloopUPV-H8/control-station/second-version

[Control Station] Second version

Author: msanlli

backend/cmd/main.go

@@ -63,6 +63,7 @@ var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
 var enableSNTP = flag.Bool("sntp", false, "enables a simple SNTP server on port 123")
 var networkDevice = flag.Int("dev", -1, "index of the network device to use, overrides device prompt")
 var blockprofile = flag.Int("blockprofile", 0, "number of block profiles to include")
+var playbackFile = flag.String("playback", "", "")
 
 func main() {
 	flag.Parse()
@@ -244,21 +245,25 @@ func main() {
 	if err != nil {
 		panic("failed to obtain sniffer source: " + err.Error())
 	}
+
+	if *playbackFile != "" {
+		source, err = pcap.OpenOffline(*playbackFile)
+		if err != nil {
+			panic("failed to obtain sniffer source: " + err.Error())
+		}
+	}
+
 	boardIps := make([]net.IP, 0)
 	for _, board := range info.Addresses.Boards {
 		boardIps = append(boardIps, board)
 	}
-	err = source.SetBPFFilter(getFilter(boardIps, info.Addresses.Backend, info.Ports.UDP, info.Ports.TcpClient, info.Ports.TcpServer))
+	filter := getFilter(boardIps, info.Addresses.Backend, info.Ports.UDP)
+	trace.Warn().Str("filter", filter).Msg("filter")
+	err = source.SetBPFFilter(filter)
 	if err != nil {
 		panic("failed to compile bpf filter")
 	}
-	go func() {
-		sniffer := sniffer.New(source, &layers.LayerTypeEthernet, trace.Logger)
-		for {
-			errChan := transp.HandleSniffer(sniffer)
-			trace.Error().Stack().Err(<-errChan).Msg("sniffer crashed, restarting...")
-		}
-	}()
+	go transp.HandleSniffer(sniffer.New(source, &layers.LayerTypeEthernet, trace.Logger))
 
 	// <--- http server --->
 	podDataHandle, err := h.HandleDataJSON("podData.json", pod_data.GetDataOnlyPodData(podData))
@@ -534,15 +539,11 @@ func getOps(units utils.Units) data.ConversionDescriptor {
 	return output
 }
 
-func getFilter(boardAddrs []net.IP, backendAddr net.IP, udpPort uint16, tcpClientPort uint16, tcpServerPort uint16) string {
+func getFilter(boardAddrs []net.IP, backendAddr net.IP, udpPort uint16) string {
 	ipipFilter := getIPIPfilter()
-	udpFilter := getUDPFilter(boardAddrs, udpPort)
-	tcpFilter := getTCPFilter(boardAddrs, tcpServerPort, tcpClientPort)
-	// noBackend := "not host 192.168.0.9"
-
-	// filter := fmt.Sprintf("((%s) or (%s) or (%s)) and (%s)", ipipFilter, udpFilter, tcpFilter, noBackend)
+	udpFilter := getUDPFilter(boardAddrs, backendAddr, udpPort)
 
-	filter := fmt.Sprintf("(%s) or (%s) or (%s)", ipipFilter, udpFilter, tcpFilter)
+	filter := fmt.Sprintf("(%s) or (%s)", ipipFilter, udpFilter)
 
 	trace.Trace().Any("addrs", boardAddrs).Str("filter", filter).Msg("new filter")
 	return filter
@@ -552,35 +553,13 @@ func getIPIPfilter() string {
 	return "ip[9] == 4"
 }
 
-func getUDPFilter(addrs []net.IP, port uint16) string {
+func getUDPFilter(addrs []net.IP, backendAddr net.IP, port uint16) string {
 	udpPort := fmt.Sprintf("udp port %d", port)
 	udpAddrs := common.Map(addrs, func(addr net.IP) string {
 		return fmt.Sprintf("(src host %s)", addr)
 	})
 
 	udpAddrsStr := strings.Join(udpAddrs, " or ")
 
-	return fmt.Sprintf("(%s) and (%s)", udpPort, udpAddrsStr)
-}
-
-func getTCPFilter(addrs []net.IP, serverPort uint16, clientPort uint16) string {
-	ports := fmt.Sprintf("tcp port %d or %d", serverPort, clientPort)
-	notSynFinRst := "tcp[tcpflags] & (tcp-fin | tcp-syn | tcp-rst) == 0"
-	notJustAck := "tcp[tcpflags] | tcp-ack != 16"
-	nonZeroPayload := "tcp[tcpflags] & tcp-push != 0"
-
-	srcAddresses := common.Map(addrs, func(addr net.IP) string {
-		return fmt.Sprintf("(src host %s)", addr)
-	})
-
-	srcAddressesStr := strings.Join(srcAddresses, " or ")
-
-	dstAddresses := common.Map(addrs, func(addr net.IP) string {
-		return fmt.Sprintf("(dst host %s)", addr)
-	})
-
-	dstAddressesStr := strings.Join(dstAddresses, " or ")
-
-	filter := fmt.Sprintf("(%s) and (%s) and (%s) and (%s) and (%s) and (%s)", ports, notSynFinRst, notJustAck, nonZeroPayload, srcAddressesStr, dstAddressesStr)
-	return filter
+	return fmt.Sprintf("(%s) and (%s) and (dst host %s)", udpPort, udpAddrsStr, backendAddr)
 }

backend/internal/common/moving_average.go

@@ -15,6 +15,10 @@ func NewMovingAverage[N Numeric](order uint) *MovingAverage[N] {
 }
 
 func (avg *MovingAverage[N]) Add(value N) N {
+	if avg == nil {
+		return 0
+	}
+
 	if avg.length < avg.Order() {
 		avg.addElem(value)
 	} else {
@@ -38,6 +42,10 @@ func (avg *MovingAverage[N]) Order() int {
 }
 
 func (avg *MovingAverage[N]) Resize(order uint) N {
+	if avg == nil {
+		return 0
+	}
+
 	if order > uint(avg.Order()) {
 		avg.Grow(order - uint(avg.Order()))
 	} else if order < uint(avg.Order()) {

backend/pkg/transport/constructor.go

@@ -11,14 +11,17 @@ import (
 )
 
 func NewTransport(baseLogger zerolog.Logger) *Transport {
-	return &Transport{
+	transport := &Transport{
 		connectionsMx: &sync.Mutex{},
 		connections:   make(map[abstraction.TransportTarget]net.Conn),
 		idToTarget:    make(map[abstraction.PacketId]abstraction.TransportTarget),
 		ipToTarget:    make(map[string]abstraction.TransportTarget),
 
-		logger: baseLogger,
+		logger:  baseLogger,
+		errChan: make(chan error, 100),
 	}
+	go transport.consumeErrors()
+	return transport
 }
 
 func (transport *Transport) WithDecoder(decoder *presentation.Decoder) *Transport {

backend/pkg/transport/messages.go

@@ -9,6 +9,8 @@ import (
 const (
 	// PacketEvent is triggered when a packet is sent or received
 	PacketEvent abstraction.TransportEvent = "packet"
+	// ErrorEvent is triggered when an error arises somewhere
+	ErrorEvent abstraction.TransportEvent = "error"
 	// FileWriteEvent is used to request a file write through tftp
 	FileWriteEvent abstraction.TransportEvent = "file-push"
 	// FileReadEvent is used to request a file read through tftp

backend/pkg/transport/network/sniffer/decoder.go

@@ -34,14 +34,14 @@ func (dec *decoder) IPv4() layers.IPv4 {
 	return dec.ipipv4
 }
 
-func (dec *decoder) TCP() layers.TCP {
-	return dec.tcp
-}
-
 func (dec *decoder) UDP() layers.UDP {
 	return dec.udp
 }
 
+func (dec *decoder) TCP() layers.TCP {
+	return dec.tcp
+}
+
 func (dec *decoder) Payload() []byte {
 	return dec.payload
 }

backend/pkg/transport/network/sniffer/sniffer.go

@@ -76,6 +76,7 @@ func (sniffer *Sniffer) ReadNext() (network.Payload, error) {
 	gotPorts := false
 	gotIp := false
 	layerArr := zerolog.Arr()
+	payloadData := []byte{}
 	for _, layer := range packetLayers {
 		if !gotIp && layer == layers.LayerTypeIPv4 {
 			ip := sniffer.decoder.IPv4()
@@ -97,6 +98,10 @@ func (sniffer *Sniffer) ReadNext() (network.Payload, error) {
 			}
 		}
 
+		if layer == gopacket.LayerTypePayload {
+			payloadData = sniffer.decoder.Payload()
+		}
+
 		layerArr = layerArr.Str(layer.String())
 	}
 
@@ -123,7 +128,7 @@ func (sniffer *Sniffer) ReadNext() (network.Payload, error) {
 
 	return network.Payload{
 		Socket:    socket,
-		Data:      sniffer.decoder.Payload(),
+		Data:      payloadData,
 		Timestamp: captureInfo.Timestamp,
 	}, nil
 }

backend/pkg/transport/network/tcp/config.go

@@ -37,16 +37,20 @@ func NewClientConfig(laddr net.Addr) ClientConfig {
 type backoffFunction = func(int) time.Duration
 
 const (
-	defaultBackoffMin time.Duration = 100 * time.Millisecond
-	defaultBackoffExp float64       = 1.5
-	defaultBackoffMax time.Duration = 5 * time.Second
+	defaultBackoffMin          time.Duration = 100 * time.Millisecond
+	defaultBackoffExp          float64       = 1.5
+	defaultBackoffMax          time.Duration = 5 * time.Second
+	defaultBackoffRetriesClamp int           = 12
 )
 
 // NewExponentialBackoff returns an exponential backoff function with the given paramenters.
 //
 // It follows this formula: delay = (min * (exp ^ n); delay < max ? delay : max
 func NewExponentialBackoff(min time.Duration, exp float64, max time.Duration) backoffFunction {
 	return func(n int) time.Duration {
+		if n > defaultBackoffRetriesClamp {
+			n = defaultBackoffRetriesClamp
+		}
 		curr := time.Duration(float64(min) * math.Trunc(math.Pow(exp, float64(n))))
 		if curr >= max {
 			return max

backend/pkg/transport/notifications.go

@@ -1,8 +1,9 @@
 package transport
 
 import (
-	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
 	"time"
+
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
 )
 
 // PacketNotification notifies of an incoming message
@@ -26,3 +27,18 @@ func NewPacketNotification(packet abstraction.Packet, from string, to string, ti
 func (notification PacketNotification) Event() abstraction.TransportEvent {
 	return PacketEvent
 }
+
+// ErrorNotification notifies of an error that arised
+type ErrorNotification struct {
+	Err error
+}
+
+func NewErrorNotification(err error) ErrorNotification {
+	return ErrorNotification{
+		Err: err,
+	}
+}
+
+func (notification ErrorNotification) Event() abstraction.TransportEvent {
+	return ErrorEvent
+}

backend/pkg/transport/packet/protection/packet.go

@@ -140,6 +140,19 @@ type Timestamp struct {
 	Year    uint16 `json:"year"`
 }
 
+func NowTimestamp() *Timestamp {
+	t := time.Now()
+	return &Timestamp{
+		Counter: 0,
+		Second:  uint8(t.Second()),
+		Minute:  uint8(t.Minute()),
+		Hour:    uint8(t.Hour()),
+		Day:     uint8(t.Day()),
+		Month:   uint8(t.Month()),
+		Year:    uint16(t.Year()),
+	}
+}
+
 func decodeTimestamp(reader io.Reader, endianness binary.ByteOrder) (*Timestamp, error) {
 	packet := new(Timestamp)
 	err := binary.Read(reader, endianness, packet)

backend/pkg/transport/presentation/decoder.go

@@ -72,7 +72,8 @@ func (decoder *Decoder) DecodeNext(reader io.Reader) (abstraction.Packet, error)
 	dec, ok := decoder.idToDecoder[id]
 	if !ok {
 		decoder.logger.Warn().Uint16("id", uint16(id)).Msg("no decoder set")
-		return nil, ErrUnexpectedId{Id: id}
+		err := ErrUnexpectedId{Id: id}
+		return nil, err
 	}
 
 	decoder.logger.Debug().Uint16("id", uint16(id)).Type("decoder", dec).Msg("decoding")