Merge pull request #660 from nainiket-d/master

logging total network bytes sent/received per host

psiphon/server/network_bandwidth_linux.go

@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2023, Psiphon Inc.
+ * All rights reserved.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package server
+
+import (
+	"bufio"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
+)
+
+func getNetworkBytesTransferred() (int64, int64, error) {
+	file, err := os.Open("/proc/net/dev")
+	if err != nil {
+		return 0, 0, errors.Trace(err)
+	}
+
+	defer file.Close()
+
+	var totalNetworkBytesReceived, totalNetworkBytesSent int64
+	scanner := bufio.NewScanner(file)
+
+	// Parsing based on the formats used by dev_seq_show and dev_seq_printf_stats:
+	// https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/tree/net/core/net-procfs.c#n105
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		fields := strings.Fields(line)
+
+		// Skip header lines, loopback interface and tunnel interface
+		if len(fields) < 17 || fields[0] == "Inter-|" || fields[0] == "face" ||
+			strings.HasPrefix(fields[0], "lo") || strings.HasPrefix(fields[0], "tun") ||
+			strings.HasPrefix(fields[0], "ipsec") || strings.HasPrefix(fields[0], "ppp") {
+			continue
+		}
+
+		// Parse received bytes
+		receivedNetworkBytes, err := strconv.ParseInt(fields[1], 10, 64)
+		if err != nil {
+			return 0, 0, errors.Trace(err)
+		}
+
+		// Parse sent bytes
+		sentNetworkBytes, err := strconv.ParseInt(fields[9], 10, 64)
+		if err != nil {
+			return 0, 0, errors.Trace(err)
+		}
+
+		totalNetworkBytesReceived += receivedNetworkBytes
+		totalNetworkBytesSent += sentNetworkBytes
+	}
+
+	if scanner.Err() != nil {
+		return 0, 0, errors.Trace(scanner.Err())
+	}
+
+	return totalNetworkBytesReceived, totalNetworkBytesSent, nil
+}

psiphon/server/network_bandwidth_other.go

@@ -0,0 +1,28 @@
+//go:build !linux
+// +build !linux
+
+/*
+ * Copyright (c) 2023, Psiphon Inc.
+ * All rights reserved.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package server
+
+func getNetworkBytesTransferred() (int64, int64, error) {
+	// No-op on this platform
+	return 0, 0, nil
+}

psiphon/server/services.go

@@ -163,12 +163,36 @@ func RunServices(configJSON []byte) (retErr error) {
 			waitGroup.Done()
 			ticker := time.NewTicker(time.Duration(config.LoadMonitorPeriodSeconds) * time.Second)
 			defer ticker.Stop()
+
+			logNetworkBytes := true
+
+			previousNetworkBytesReceived, previousNetworkBytesSent, err := getNetworkBytesTransferred()
+			if err != nil {
+				log.WithTraceFields(LogFields{"error": errors.Trace(err)}).Error("failed to get initial network bytes transferred")
+				logNetworkBytes = false
+			}
+
 			for {
 				select {
 				case <-shutdownBroadcast:
 					return
 				case <-ticker.C:
-					logServerLoad(support)
+					var networkBytesReceived, networkBytesSent int64
+
+					if logNetworkBytes {
+						currentNetworkBytesReceived, currentNetworkBytesSent, err := getNetworkBytesTransferred()
+						if err != nil {
+							log.WithTraceFields(LogFields{"error": errors.Trace(err)}).Error("failed to get current network bytes transferred")
+							logNetworkBytes = false
+						}
+
+						networkBytesReceived = currentNetworkBytesReceived - previousNetworkBytesReceived
+						networkBytesSent = currentNetworkBytesSent - previousNetworkBytesSent
+
+						previousNetworkBytesReceived, previousNetworkBytesSent = currentNetworkBytesReceived, currentNetworkBytesSent
+					}
+
+					logServerLoad(support, logNetworkBytes, networkBytesReceived, networkBytesSent)
 				}
 			}
 		}()
@@ -284,7 +308,7 @@ loop:
 			case signalProcessProfiles <- struct{}{}:
 			default:
 			}
-			logServerLoad(support)
+			logServerLoad(support, false, 0, 0)
 
 		case <-systemStopSignal:
 			log.WithTrace().Info("shutdown by system")
@@ -362,12 +386,17 @@ func outputProcessProfiles(config *Config, filenameSuffix string) {
 	}
 }
 
-func logServerLoad(support *SupportServices) {
+func logServerLoad(support *SupportServices, logNetworkBytes bool, networkBytesReceived int64, networkBytesSent int64) {
 
 	serverLoad := getRuntimeMetrics()
 
 	serverLoad["event_name"] = "server_load"
 
+	if logNetworkBytes {
+		serverLoad["network_bytes_received"] = networkBytesReceived
+		serverLoad["network_bytes_sent"] = networkBytesSent
+	}
+
 	establishTunnels, establishLimitedCount :=
 		support.TunnelServer.GetEstablishTunnelsMetrics()
 	serverLoad["establish_tunnels"] = establishTunnels