Inconsistent UDP Message Reception

[radovan-kosc-vissim]

Good morning

I am trying to receive UDP messages using UDP Input. Interestingly, while messages from one application successfully pass through Fluent Bit, I am encountering an issue where Fluent Bit does not accept messages from another application. This is what the unsuccesfull output is:

[2024/04/29 16:42:59] [trace] [input:udp:udp.0 at D:\a\fluent-bit\fluent-bit\plugins\in_udp\udp_conn.c:387] read()=11 pre_len=0 now_len=11
[2024/04/29 16:42:59] [debug] [input chunk] skip ingesting data with 0 bytes

Upon further analysis using Wireshark, there is a slight distinction between the packets that successfully traverse Fluent Bit and those which don't. Specifically, it appears that the messages that successfully pass through Fluent Bit do not contain the source port information in the data byte.

Is it possible that when a UDP message reaches Fluent Bit, Fluent Bit itself attempts to contact back the source or initiates a response? Perhaps, due to the absence of port information in the first application's messages, Fluent Bit allows them to pass through. However, when it attempts to respond to messages from the second application, which include port information, it encounters difficulties, resulting in the messages being marked as empty.

Thanks for any insights to this matter

[patrick-stephens]

It looks more like it's reading a field in the message and getting a size of 0 so skipping it - my guess is adding a source port is moving an offset. It should really cope with it fine though.

[radovan-kosc-vissim]

I'd like to rephrase.

Fluent Bit is able to receive the messages when the application does not listen on the source UDP port. When the application where the source listens on the UDP the messages do not come through with the error I mentioned in my first post.

Could you maybe give me more details about how UDP Input works or point me to the code where its implemented?

[ghost]

I am not able to receive any message on UDP input when I run this code:

package main

import (
	"fmt"
	"log"
	"net"
	"time"
)

func main() {
	// https://docs.fluentbit.io/manual/pipeline/inputs/udp#fluent-bit.conf
	// fluent-bit -i udp -pformat=none -o stdout -f 1
	dest := net.UDPAddr{
		Port: 5170,
		IP:   net.ParseIP("127.0.0.1"),
	}

	go ListeningAndSending(dest)
	go JustSending(dest)

	// Keep the main function alive indefinitely
	select {}
}

func JustSending(dest net.UDPAddr) {
	conn, err := net.DialUDP("udp", nil, &dest)
	if err != nil {
		log.Fatalf("Error dialing UDP: %v", err)
	}
	defer conn.Close()

	// Get the client's port
	clientAddr := conn.LocalAddr().(*net.UDPAddr)
	log.Printf("Client port: %d\n", clientAddr.Port)

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	counter := 0
	for range ticker.C {
		counter++
		// msg := fmt.Sprintf("Just Sending! %d", counter)
		msg := fmt.Sprintf("{\"counter\": %d, \"method\": \"JustSending\"}", counter)
		bytes := []byte(msg)
		_, err = conn.Write(bytes)
		if err != nil {
			log.Fatalf("Error writing to UDP: %v", err)
		}
		log.Println("Message from JustSending")
	}
}

func ListeningAndSending(dest net.UDPAddr) {
	addr := net.UDPAddr{
		// Port: 5170,
		Port: 12345,
		IP:   net.ParseIP("127.0.0.1"),
	}
	conn, err := net.ListenUDP("udp", &addr)
	if err != nil {
		log.Fatalf("Error listening UDP: %v", err)
	}
	defer conn.Close()

	log.Printf("Destination: %s\n", dest.String())

	// Get the client's port
	clientAddr := conn.LocalAddr().(*net.UDPAddr)
	log.Printf("Client port: %d\n", clientAddr.Port)

	go func() {
		buffer := make([]byte, 1024)
		for {
			n, addr, err := conn.ReadFromUDP(buffer)
			if err != nil {
				log.Fatalf("Error reading from UDP: %v", err)
			}
			log.Printf("Received %s from %s\n", string(buffer[:n]), addr)
		}
	}()

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	counter := 0
	for range ticker.C {
		counter++
		// msg := fmt.Sprintf("Listening And Sending! %d", counter)
		msg := fmt.Sprintf("{\"counter\": %d, \"method\": \"ListeningAndSending\"}", counter)
		bytes := []byte(msg)
		_, err = conn.WriteToUDP(bytes, &dest)
		if err != nil {
			log.Fatalf("Error writing to UDP: %v", err)
		}
		log.Println("Message sent from ListeningAndSending")
	}
}

Whent I start fluent-bit with format=none (fluent-bit -i udp -pformat=none -o stdout -f 1) I do not receieve any messages.

However, when I start fluentbit with format=json: fluent-bit -i udp -pformat=json -o stdout -f 1 messages are received.

We are using UDP input for some legacy reporting mechanism and then simple lua script to translate received data to metrics.