config/monitoring/pinger/main.go

package main

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"regexp"
	"strconv"
	"time"

	"github.com/nxadm/tail"
	ping "github.com/prometheus-community/pro-bing"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

type IperfResult struct {
	Start struct {
		Version     string  `json:"version"`
		SystemInfo  string  `json:"system_info"`
		ConnectTime float64 `json:"connect_time"`
	} `json:"start"`
	End struct {
		SumSent struct {
			Seconds     float64 `json:"seconds"`
			Bytes       int64   `json:"bytes"`
			BitsPerSec  float64 `json:"bits_per_second"`
			Retransmits int     `json:"retransmits"`
		} `json:"sum_sent"`
		SumReceived struct {
			Seconds    float64 `json:"seconds"`
			Bytes      int64   `json:"bytes"`
			BitsPerSec float64 `json:"bits_per_second"`
		} `json:"sum_received"`
		CPUUtilization struct {
			HostTotal   float64 `json:"host_total"`
			RemoteTotal float64 `json:"remote_total"`
		} `json:"cpu_utilization_percent"`
	} `json:"end"`
	Error string `json:"error"`
}

var Error *log.Logger
var Warning *log.Logger

var (
	pingRttHistogram = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "ping_rtt_seconds",
			Help:    "Round-trip time histogram in seconds",
			Buckets: []float64{.001, .002, .005, .01, .02, .05, .1, .2, .5, 1},
		},
		[]string{"target", "name"},
	)

	iperfBandwidthGauge = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "iperf_bandwidth_bits_per_second",
			Help: "Bandwidth measured by iperf3 in bits per second",
		},
		[]string{"direction", "server"},
	)

	httpRttHistogram = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "http_rtt_seconds",
			Help:    "HTTP request round-trip time histogram in seconds",
			Buckets: []float64{.001, .002, .005, .01, .02, .05, .1, .2, .5, 1},
		},
		[]string{"target", "name"},
	)
)

var iperfServer = "138.199.199.16"
var iperfUploadPort = 3332
var iperfDownloadPort = 3333

func init() {
	log.Printf("Initializing application...")
	log.SetFlags(log.Lmicroseconds | log.Lshortfile)
	Error = log.New(io.MultiWriter(os.Stderr, os.Stdout),
		fmt.Sprintf("%sERROR:%s ", "\033[0;101m", "\033[0m"),
		log.Lmicroseconds|log.Lshortfile)
	Warning = log.New(io.MultiWriter(os.Stdout),
		fmt.Sprintf("%sWarning:%s ", "\033[0;93m", "\033[0m"),
		log.Lmicroseconds|log.Lshortfile)

	// Register only the essential metrics
	prometheus.MustRegister(pingRttHistogram)

	// Register new iperf3 metrics
	prometheus.MustRegister(iperfBandwidthGauge)

	// Register HTTP metrics
	prometheus.MustRegister(httpRttHistogram)

	log.Printf("Registered all Prometheus metrics")
}

type PingTarget struct {
	IP   string
	Name string
}

func updateMetrics(stats *ping.Statistics, name string) {
	target := stats.Addr

	if stats.PacketsRecv > 0 {
		histogram := pingRttHistogram.WithLabelValues(target, name)
		histogram.Observe(stats.AvgRtt.Seconds())
		//log.Printf("Ping metrics for %s (%s): RTT=%v, PacketsRecv=%d/%d",
		//name, target, stats.AvgRtt, stats.PacketsRecv, stats.PacketsSent)
	} else {
		log.Printf("No packets received for %s (%s)", name, target)
	}
}

// IperfTest represents a runnable iperf3 test configuration
type IperfTest struct {
	ServerHost   string
	UploadPort   int
	DownloadPort int
}

// NewIperfTest creates a new IperfTest with default values
func NewIperfTest(serverHost string, uploadPort, downloadPort int) *IperfTest {
	return &IperfTest{
		ServerHost:   serverHost,
		UploadPort:   uploadPort,
		DownloadPort: downloadPort,
	}
}

var bitrateRegex = regexp.MustCompile(`(\d+) Mbits/sec`)

// runTest executes an iperf3 test with the given direction
func (t *IperfTest) runTest(reverse bool) error {
	direction := "upload"
	port := t.UploadPort
	if reverse {
		direction = "download"
		port = t.DownloadPort
	}

	log.Printf("Starting iperf3 %s test to %s:%d", direction, t.ServerHost, port)

	// Check if iperf3 is installed
	if _, err := exec.LookPath("iperf3"); err != nil {
		return fmt.Errorf("iperf3 not found: %v", err)
	}

	title := "upload"
	if reverse {
		title = "download"
	}

	logFile := fmt.Sprintf("iperf3-%s-%s.log", direction, t.ServerHost)
	_, err := os.Stat(logFile)
	if err == nil {
		log.Printf("Removing existing log file %s", logFile)
		os.Remove(logFile)
	}
	log.Printf("Logging iperf3 %s test to %s", direction, logFile)

	args := []string{
		"-c", t.ServerHost,
		"-p", fmt.Sprintf("%d", port),
		"-Z",
		"-T", title,
		"-t", "0",
		"-b", "600M", // Limit bandwidth to 200 Mbits
		"-f", "m",
		"-i", "1",
		"--logfile", logFile,
	}

	if reverse {
		args = append(args, "-R")
	}

	log.Printf("Running iperf3 %s test with args: %v", direction, args)

	cmd := exec.Command("iperf3", args...)
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		return fmt.Errorf("failed to start iperf3: %v", err)
	}

	itail, err := tail.TailFile(
		logFile, tail.Config{Follow: true, ReOpen: true})
	if err != nil {
		return fmt.Errorf("failed to tail iperf3 log file: %v", err)
	}

	for line := range itail.Lines {
		bitrate := bitrateRegex.FindStringSubmatch(line.Text)
		if len(bitrate) > 1 {
			mbps, err := strconv.ParseFloat(bitrate[1], 64)
			if err != nil {
				log.Printf("failed to parse bitrate: %v", err)
			}

			log.Printf("iperf3 %s bandwidth: %.2f Mbps", direction, mbps)
			if reverse {
				iperfBandwidthGauge.WithLabelValues(direction, t.ServerHost).Set(mbps)
			} else {
				iperfBandwidthGauge.WithLabelValues(direction, t.ServerHost).Set(mbps)
			}
		}
	}

	if err := cmd.Wait(); err != nil {
		return fmt.Errorf("iperf3 %s test failed: %v",
			direction, err)
	}
	log.Printf("iperf3 %s test completed successfully", direction)

	return nil
}

// Start begins continuous iperf3 testing in both directions
func (t *IperfTest) Start() {
	log.Printf("Starting continuous iperf3 tests to %s", t.ServerHost)

	// Start upload test goroutine
	go func() {
		for {
			if err := t.runTest(false); err != nil {
				Error.Printf("Upload test failed: %v", err)
			}
			time.Sleep(1 * time.Second)
			log.Printf("Restarting upload test immediately")
		}
	}()

	// Start download test goroutine
	go func() {
		for {
			if err := t.runTest(true); err != nil {
				Error.Printf("Download test failed: %v", err)
			}
			time.Sleep(1 * time.Second)
			log.Printf("Restarting download test immediately")
		}
	}()
}

type HttpTarget struct {
	URL  string
	Name string
}

func probeHTTP(target HttpTarget) {
	log.Printf("Starting HTTP probe for %s (%s)", target.Name, target.URL)
	parsedURL, err := url.Parse(target.URL)
	if err != nil {
		Error.Printf("Invalid URL %s: %v", target.URL, err)
		return
	}

	client := &http.Client{
		Timeout: 20 * time.Second,
	}

	for {
		start := time.Now()
		log.Printf("Sending HTTP request to %s", target.URL)
		resp, err := client.Get(target.URL)
		duration := time.Since(start)

		if err != nil {
			Error.Printf("HTTP probe to %s failed: %v (took %v)", target.URL, err, duration)
		} else {
			httpRttHistogram.WithLabelValues(parsedURL.Host, target.Name).Observe(duration.Seconds())
			log.Printf("HTTP probe to %s succeeded in %v", target.URL, duration)
			resp.Body.Close()
		}

		log.Printf("Sleeping for 10 seconds before next HTTP probe to %s", target.URL)
		time.Sleep(10 * time.Second)
	}
}

func main() {
	log.Printf("Starting network monitoring application")

	targets := []PingTarget{
		{IP: "8.8.8.8", Name: "google-dns-1"},
		{IP: "1.1.1.1", Name: "cloudflare-dns-1"},
		{IP: "8.8.4.4", Name: "google-dns-2"},
		{IP: "1.0.0.1", Name: "cloudflare-dns-2"},
		{IP: "208.67.222.222", Name: "opendns-1"},
		{IP: "208.67.220.220", Name: "opendns-2"},
		{IP: "192.168.1.254", Name: "router"},
	}
	log.Printf("Configured ping targets: %v", targets)

	// Add HTTP targets
	httpTargets := []HttpTarget{
		{URL: "http://192.168.1.254", Name: "router-http"},
	}
	log.Printf("Configured HTTP targets: %v", httpTargets)

	// Start HTTP probes
	for _, target := range httpTargets {
		go probeHTTP(target)
	}

	//Start Prometheus HTTP server
	go func() {
		http.Handle("/metrics", promhttp.Handler())
		if err := http.ListenAndServe(":2112", nil); err != nil {
			Error.Printf("starting prometheus listener failed: %v", err)
			os.Exit(1)
		}
	}()

	pingers := make([]*ping.Pinger, len(targets))
	for i, target := range targets {
		pinger, err := ping.NewPinger(target.IP)
		if err != nil {
			Error.Printf("new pinger for %s failed: %v", target.IP, err)
			return
		}
		//pinger.SetPrivileged(true)
		pinger.RecordRtts = false
		pinger.RecordTTLs = false
		pingers[i] = pinger
		go func(t PingTarget) {
			log.Printf("Starting pinger for %s (%s)", t.Name, t.IP)
			err = pinger.Run()
			if err != nil {
				Error.Printf("pinger for %s (%s) failed: %v", t.Name, t.IP, err)
				return
			}
		}(target)
	}

	// Initialize and start iperf3 testing with separate ports
	iperfTest := NewIperfTest(iperfServer, iperfUploadPort, iperfDownloadPort)
	iperfTest.Start()

	log.Printf("All monitoring services started, entering main loop")
	for {
		time.Sleep(1 * time.Second)
		for i, pinger := range pingers {
			if pinger != nil {
				stats := pinger.Statistics()
				updateMetrics(stats, targets[i].Name)
			}
		}
	}
}