Split regional breakdowns out of server_load logs

psiphon/server/services.go

@@ -202,6 +202,20 @@ loop:
 
 func outputProcessProfiles(config *Config) {
 
+	var memStats runtime.MemStats
+	runtime.ReadMemStats(&memStats)
+	log.WithContextFields(
+		LogFields{
+			"num_goroutine":   runtime.NumGoroutine(),
+			"alloc":           memStats.Alloc,
+			"total_alloc":     memStats.TotalAlloc,
+			"sys":             memStats.Sys,
+			"pause_total_ns":  memStats.PauseTotalNs,
+			"pause_ns":        memStats.PauseNs,
+			"num_gc":          memStats.NumGC,
+			"gc_cpu_fraction": memStats.GCCPUFraction,
+		}).Info("runtime_stats")
+
 	if config.ProcessProfileOutputDirectory != "" {
 
 		openProfileFile := func(profileName string) *os.File {
@@ -283,33 +297,34 @@ func outputProcessProfiles(config *Config) {
 
 func logServerLoad(server *TunnelServer) {
 
-	// golang runtime stats
+	protocolStats, regionStats := server.GetLoadStats()
 
-	var memStats runtime.MemStats
-	runtime.ReadMemStats(&memStats)
-	fields := LogFields{
-		"event_name":    "server_load",
-		"num_goroutine": runtime.NumGoroutine(),
-		"mem_stats": map[string]interface{}{
-			"alloc":           memStats.Alloc,
-			"total_alloc":     memStats.TotalAlloc,
-			"sys":             memStats.Sys,
-			"pause_total_ns":  memStats.PauseTotalNs,
-			"pause_ns":        memStats.PauseNs,
-			"num_gc":          memStats.NumGC,
-			"gc_cpu_fraction": memStats.GCCPUFraction,
-		},
+	serverLoad := LogFields{
+		"event_name": "server_load",
+	}
+	for protocol, stats := range protocolStats {
+		serverLoad[protocol] = stats
 	}
+	serverLoad["establish_tunnels"] = server.GetEstablishTunnels()
 
-	// tunnel server stats
+	log.LogRawFieldsWithTimestamp(serverLoad)
 
-	fields["establish_tunnels"] = server.GetEstablishTunnels()
+	for protocol, regions := range regionStats {
+		for region, stats := range regions {
 
-	for tunnelProtocol, stats := range server.GetLoadStats() {
-		fields[tunnelProtocol] = stats
-	}
+			serverRegionLoad := LogFields{
+				"event_name": "server_region_load",
+				"protocol":   protocol,
+				"region":     region,
+			}
+
+			for name, value := range stats {
+				serverRegionLoad[name] = value
+			}
 
-	log.LogRawFieldsWithTimestamp(fields)
+			log.LogRawFieldsWithTimestamp(serverRegionLoad)
+		}
+	}
 }
 
 // SupportServices carries common and shared data components

psiphon/server/tunnelServer.go

@@ -193,7 +193,7 @@ func (server *TunnelServer) Run() error {
 // broken down by protocol ("SSH", "OSSH", etc.) and type. Types of stats
 // include current connected client count, total number of current port
 // forwards.
-func (server *TunnelServer) GetLoadStats() map[string]interface{} {
+func (server *TunnelServer) GetLoadStats() (ProtocolStats, RegionStats) {
 	return server.sshServer.getLoadStats()
 }
 
@@ -443,88 +443,98 @@ func (sshServer *sshServer) unregisterEstablishedClient(sessionID string) {
 	}
 }
 
-func (sshServer *sshServer) getLoadStats() map[string]interface{} {
+type ProtocolStats map[string]map[string]int64
+type RegionStats map[string]map[string]map[string]int64
+
+func (sshServer *sshServer) getLoadStats() (ProtocolStats, RegionStats) {
 
 	sshServer.clientsMutex.Lock()
 	defer sshServer.clientsMutex.Unlock()
 
-	protocolStats := make(map[string]map[string]map[string]int64)
-
-	// Explicitly populate with zeros to get 0 counts in log messages derived from getLoadStats()
-
+	// [<protocol or ALL>][<stat name] -> count
+	protocolStats := make(ProtocolStats)
+
+	// [<protocol>][<region][<stat name] -> count
+	regionStats := make(RegionStats)
+
+	// Explicitly populate with zeros to ensure 0 counts in log messages
+	zeroStats := func() map[string]int64 {
+		stats := make(map[string]int64)
+		stats["accepted_clients"] = 0
+		stats["established_clients"] = 0
+		stats["dialing_tcp_port_forwards"] = 0
+		stats["tcp_port_forwards"] = 0
+		stats["total_tcp_port_forwards"] = 0
+		stats["udp_port_forwards"] = 0
+		stats["total_udp_port_forwards"] = 0
+		stats["tcp_port_forward_dialed_count"] = 0
+		stats["tcp_port_forward_dialed_duration"] = 0
+		stats["tcp_port_forward_failed_count"] = 0
+		stats["tcp_port_forward_failed_duration"] = 0
+		stats["tcp_port_forward_rejected_dialing_limit_count"] = 0
+		return stats
+	}
+
+	protocolStats["ALL"] = zeroStats()
 	for tunnelProtocol, _ := range sshServer.support.Config.TunnelProtocolPorts {
-		protocolStats[tunnelProtocol] = make(map[string]map[string]int64)
-		protocolStats[tunnelProtocol]["ALL"] = make(map[string]int64)
-		protocolStats[tunnelProtocol]["ALL"]["accepted_clients"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["established_clients"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["dialing_tcp_port_forwards"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["tcp_port_forwards"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["total_tcp_port_forwards"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["udp_port_forwards"] = 0
-		protocolStats[tunnelProtocol]["ALL"]["total_udp_port_forwards"] = 0
+		protocolStats[tunnelProtocol] = zeroStats()
+		regionStats[tunnelProtocol] = make(map[string]map[string]int64)
 	}
 
 	// Note: as currently tracked/counted, each established client is also an accepted client
 
 	for tunnelProtocol, regionAcceptedClientCounts := range sshServer.acceptedClientCounts {
-		total := int64(0)
 		for region, acceptedClientCount := range regionAcceptedClientCounts {
 			if acceptedClientCount > 0 {
-				if protocolStats[tunnelProtocol][region] == nil {
-					protocolStats[tunnelProtocol][region] = make(map[string]int64)
-					protocolStats[tunnelProtocol][region]["accepted_clients"] = 0
-					protocolStats[tunnelProtocol][region]["established_clients"] = 0
-					protocolStats[tunnelProtocol][region]["dialing_tcp_port_forwards"] = 0
-					protocolStats[tunnelProtocol][region]["tcp_port_forwards"] = 0
-					protocolStats[tunnelProtocol][region]["total_tcp_port_forwards"] = 0
-					protocolStats[tunnelProtocol][region]["udp_port_forwards"] = 0
-					protocolStats[tunnelProtocol][region]["total_udp_port_forwards"] = 0
+				if regionStats[tunnelProtocol][region] == nil {
+					regionStats[tunnelProtocol][region] = zeroStats()
 				}
-				protocolStats[tunnelProtocol][region]["accepted_clients"] = acceptedClientCount
-				total += acceptedClientCount
+				protocolStats["ALL"]["accepted_clients"] += acceptedClientCount
+				protocolStats[tunnelProtocol]["accepted_clients"] += acceptedClientCount
+				regionStats[tunnelProtocol][region]["accepted_clients"] += acceptedClientCount
 			}
 		}
-		protocolStats[tunnelProtocol]["ALL"]["accepted_clients"] = total
 	}
 
-	var aggregatedQualityMetrics qualityMetrics
-
 	for _, client := range sshServer.clients {
 
 		client.Lock()
 
-		for _, region := range []string{"ALL", client.geoIPData.Country} {
-
-			if protocolStats[client.tunnelProtocol][region] == nil {
-				protocolStats[client.tunnelProtocol][region] = make(map[string]int64)
-				protocolStats[client.tunnelProtocol][region]["accepted_clients"] = 0
-				protocolStats[client.tunnelProtocol][region]["established_clients"] = 0
-				protocolStats[client.tunnelProtocol][region]["dialing_tcp_port_forwards"] = 0
-				protocolStats[client.tunnelProtocol][region]["tcp_port_forwards"] = 0
-				protocolStats[client.tunnelProtocol][region]["total_tcp_port_forwards"] = 0
-				protocolStats[client.tunnelProtocol][region]["udp_port_forwards"] = 0
-				protocolStats[client.tunnelProtocol][region]["total_udp_port_forwards"] = 0
-			}
+		tunnelProtocol := client.tunnelProtocol
+		region := client.geoIPData.Country
+
+		if regionStats[tunnelProtocol][region] == nil {
+			regionStats[tunnelProtocol][region] = zeroStats()
+		}
 
-			// Note: can't sum trafficState.peakConcurrentPortForwardCount to get a global peak
-			protocolStats[client.tunnelProtocol][region]["established_clients"] += 1
+		// Note: can't sum trafficState.peakConcurrentPortForwardCount to get a global peak
 
-			protocolStats[client.tunnelProtocol][region]["dialing_tcp_port_forwards"] += client.tcpTrafficState.concurrentDialingPortForwardCount
-			protocolStats[client.tunnelProtocol][region]["tcp_port_forwards"] += client.tcpTrafficState.concurrentPortForwardCount
-			protocolStats[client.tunnelProtocol][region]["total_tcp_port_forwards"] += client.tcpTrafficState.totalPortForwardCount
-			// client.udpTrafficState.concurrentDialingPortForwardCount isn't meaningful
-			protocolStats[client.tunnelProtocol][region]["udp_port_forwards"] += client.udpTrafficState.concurrentPortForwardCount
-			protocolStats[client.tunnelProtocol][region]["total_udp_port_forwards"] += client.udpTrafficState.totalPortForwardCount
+		stats := []map[string]int64{
+			protocolStats["ALL"],
+			protocolStats[tunnelProtocol],
+			regionStats[tunnelProtocol][region]}
 
+		for _, stat := range stats {
+
+			stat["established_clients"] += 1
+
+			stat["dialing_tcp_port_forwards"] += client.tcpTrafficState.concurrentDialingPortForwardCount
+			stat["tcp_port_forwards"] += client.tcpTrafficState.concurrentPortForwardCount
+			stat["total_tcp_port_forwards"] += client.tcpTrafficState.totalPortForwardCount
+			// client.udpTrafficState.concurrentDialingPortForwardCount isn't meaningful
+			stat["udp_port_forwards"] += client.udpTrafficState.concurrentPortForwardCount
+			stat["total_udp_port_forwards"] += client.udpTrafficState.totalPortForwardCount
+
+			stat["tcp_port_forward_dialed_count"] += client.qualityMetrics.tcpPortForwardDialedCount
+			stat["tcp_port_forward_dialed_duration"] +=
+				int64(client.qualityMetrics.tcpPortForwardDialedDuration / time.Millisecond)
+			stat["tcp_port_forward_failed_count"] += client.qualityMetrics.tcpPortForwardFailedCount
+			stat["tcp_port_forward_failed_duration"] +=
+				int64(client.qualityMetrics.tcpPortForwardFailedDuration / time.Millisecond)
+			stat["tcp_port_forward_rejected_dialing_limit_count"] +=
+				client.qualityMetrics.tcpPortForwardRejectedDialingLimitCount
 		}
 
-		aggregatedQualityMetrics.tcpPortForwardDialedCount += client.qualityMetrics.tcpPortForwardDialedCount
-		aggregatedQualityMetrics.tcpPortForwardDialedDuration +=
-			client.qualityMetrics.tcpPortForwardDialedDuration / time.Millisecond
-		aggregatedQualityMetrics.tcpPortForwardFailedCount += client.qualityMetrics.tcpPortForwardFailedCount
-		aggregatedQualityMetrics.tcpPortForwardFailedDuration +=
-			client.qualityMetrics.tcpPortForwardFailedDuration / time.Millisecond
-		aggregatedQualityMetrics.tcpPortForwardRejectedDialingLimitCount += client.qualityMetrics.tcpPortForwardRejectedDialingLimitCount
 		client.qualityMetrics.tcpPortForwardDialedCount = 0
 		client.qualityMetrics.tcpPortForwardDialedDuration = 0
 		client.qualityMetrics.tcpPortForwardFailedCount = 0
@@ -534,36 +544,7 @@ func (sshServer *sshServer) getLoadStats() map[string]interface{} {
 		client.Unlock()
 	}
 
-	// Calculate and report totals across all protocols. It's easier to do this here
-	// than futher down the stats stack. Also useful for glancing at log files.
-
-	allProtocolsStats := make(map[string]int64)
-	allProtocolsStats["accepted_clients"] = 0
-	allProtocolsStats["established_clients"] = 0
-	allProtocolsStats["dialing_tcp_port_forwards"] = 0
-	allProtocolsStats["tcp_port_forwards"] = 0
-	allProtocolsStats["total_tcp_port_forwards"] = 0
-	allProtocolsStats["udp_port_forwards"] = 0
-	allProtocolsStats["total_udp_port_forwards"] = 0
-	allProtocolsStats["tcp_port_forward_dialed_count"] = aggregatedQualityMetrics.tcpPortForwardDialedCount
-	allProtocolsStats["tcp_port_forward_dialed_duration"] = int64(aggregatedQualityMetrics.tcpPortForwardDialedDuration)
-	allProtocolsStats["tcp_port_forward_failed_count"] = aggregatedQualityMetrics.tcpPortForwardFailedCount
-	allProtocolsStats["tcp_port_forward_failed_duration"] = int64(aggregatedQualityMetrics.tcpPortForwardFailedDuration)
-	allProtocolsStats["tcp_port_forward_rejected_dialing_limit_count"] = aggregatedQualityMetrics.tcpPortForwardRejectedDialingLimitCount
-
-	for _, stats := range protocolStats {
-		for name, value := range stats["ALL"] {
-			allProtocolsStats[name] += value
-		}
-	}
-
-	loadStats := make(map[string]interface{})
-	loadStats["ALL"] = allProtocolsStats
-	for tunnelProtocol, stats := range protocolStats {
-		loadStats[tunnelProtocol] = stats
-	}
-
-	return loadStats
+	return protocolStats, regionStats
 }
 
 func (sshServer *sshServer) resetAllClientTrafficRules() {