Merge pull request #33 from risingglory/fix-30day-query-chunking

Fix 30-day query failures with smart chunking and sampling

Author: rajsinghtech

backend/internal/handlers/handlers.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/rajsinghtech/tsflow/backend/internal/services"
+	tailscale "tailscale.com/client/tailscale/v2"
 )
 
 type Handlers struct {
@@ -107,8 +108,12 @@ func (h *Handlers) GetNetworkLogs(c *gin.Context) {
 	}
 
 	duration := et.Sub(st)
-	if duration > 24*time.Hour {
-		chunks, err := h.tailscaleService.GetNetworkLogsChunkedParallel(start, end, 6*time.Hour, 3)
+	// Use chunking for queries longer than 7 days to prevent response size issues
+	if duration > 7*24*time.Hour {
+		// Use smaller chunks and fewer parallel requests for 30+ day queries
+		chunkSize := 24 * time.Hour // 1-day chunks to prevent timeouts
+		maxParallel := 2            // Reduce parallel requests to prevent memory issues
+		chunks, err := h.tailscaleService.GetNetworkLogsChunkedParallel(start, end, chunkSize, maxParallel)
 		if err != nil {
 			c.JSON(http.StatusInternalServerError, gin.H{
 				"error":   "Failed to fetch network logs",
@@ -119,25 +124,66 @@ func (h *Handlers) GetNetworkLogs(c *gin.Context) {
 		}
 
 		var allLogs []interface{}
+		maxLogs := 10000 // Limit total logs to prevent memory issues
+		
 		for _, chunk := range chunks {
 			if logsArray, ok := chunk.([]interface{}); ok {
+				if len(allLogs)+len(logsArray) > maxLogs {
+					// Truncate if we're approaching the limit
+					remaining := maxLogs - len(allLogs)
+					if remaining > 0 {
+						allLogs = append(allLogs, logsArray[:remaining]...)
+					}
+					break
+				}
 				allLogs = append(allLogs, logsArray...)
 			} else if logsMap, ok := chunk.(map[string]interface{}); ok {
 				if logs, exists := logsMap["logs"]; exists {
 					if logsArray, ok := logs.([]interface{}); ok {
+						if len(allLogs)+len(logsArray) > maxLogs {
+							// Truncate if we're approaching the limit
+							remaining := maxLogs - len(allLogs)
+							if remaining > 0 {
+								allLogs = append(allLogs, logsArray[:remaining]...)
+							}
+							break
+						}
 						allLogs = append(allLogs, logsArray...)
+					} else if logsArray, ok := logs.([]tailscale.NetworkFlowLog); ok {
+						// Convert []NetworkFlowLog to []interface{}
+						for _, log := range logsArray {
+							allLogs = append(allLogs, log)
+						}
 					}
 				}
 			}
 		}
-
+		
+		// If we have too many logs, sample them to prevent response size issues
+		finalLogs := allLogs
+		if len(allLogs) > 50000 {
+			// Sample every Nth log to get approximately 50,000 logs
+			sampleRate := len(allLogs) / 50000
+			if sampleRate < 1 {
+				sampleRate = 1
+			}
+			
+			sampledLogs := make([]interface{}, 0, 50000)
+			for i := 0; i < len(allLogs); i += sampleRate {
+				sampledLogs = append(sampledLogs, allLogs[i])
+			}
+			finalLogs = sampledLogs
+		}
+		
 		c.JSON(http.StatusOK, gin.H{
-			"logs": allLogs,
+			"logs": finalLogs,
 			"metadata": gin.H{
 				"chunked":     true,
 				"chunks":      len(chunks),
 				"duration":    duration.String(),
 				"totalLogs":   len(allLogs),
+				"sampled":     len(finalLogs) < len(allLogs),
+				"sampleRate":  len(allLogs) / len(finalLogs),
 			},
 		})
 		return
@@ -152,10 +198,18 @@ func (h *Handlers) GetNetworkLogs(c *gin.Context) {
 		return
 	}
 
-	log.Printf("SUCCESS GetNetworkLogs: returned logs for %s to %s", start, end)
 	c.JSON(http.StatusOK, logs)
 }
 
+// Helper function to get map keys
+func getMapKeys(m map[string]interface{}) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
 func (h *Handlers) GetNetworkMap(c *gin.Context) {
 	networkMap, err := h.tailscaleService.GetNetworkMap()
 	if err != nil {

backend/internal/services/tailscale.go

@@ -87,7 +87,7 @@ func NewTailscaleService(cfg *config.Config) *TailscaleService {
 	} else if cfg.TailscaleAPIKey != "" {
 		ts.apiKey = cfg.TailscaleAPIKey
 		ts.client = &http.Client{
-			Timeout: 5 * time.Minute,
+			Timeout: 30 * time.Minute, // Much longer timeout for large requests
 		}
 		ts.tsClient = &tailscale.Client{
 			APIKey:  cfg.TailscaleAPIKey,
@@ -96,7 +96,7 @@ func NewTailscaleService(cfg *config.Config) *TailscaleService {
 		ts.useOAuth = false
 	} else {
 		ts.client = &http.Client{
-			Timeout: 5 * time.Minute,
+			Timeout: 30 * time.Minute, // Much longer timeout for large requests
 		}
 	}
 
@@ -243,19 +243,27 @@ func (ts *TailscaleService) GetDevices() (*DevicesResponse, error) {
 }
 
 func (ts *TailscaleService) GetNetworkLogs(start, end string) (interface{}, error) {
+	// Parse time range to determine if we need chunking
+	startTime, err := time.Parse(time.RFC3339, start)
+	if err != nil {
+		return nil, fmt.Errorf("invalid start time: %w", err)
+	}
+	
+	endTime, err := time.Parse(time.RFC3339, end)
+	if err != nil {
+		return nil, fmt.Errorf("invalid end time: %w", err)
+	}
+	
+	
+	// For smaller ranges, use the original approach
 	if ts.tsClient != nil {
-		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
-		defer cancel()
-		
-		startTime, err := time.Parse(time.RFC3339, start)
-		if err != nil {
-			return nil, fmt.Errorf("invalid start time: %w", err)
-		}
-		
-		endTime, err := time.Parse(time.RFC3339, end)
-		if err != nil {
-			return nil, fmt.Errorf("invalid end time: %w", err)
+		// Use much longer timeout for larger time ranges
+		timeoutDuration := 10 * time.Minute
+		if endTime.Sub(startTime) > 7*24*time.Hour {
+			timeoutDuration = 30 * time.Minute // Much longer timeout for 30+ day queries
 		}
+		ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
+		defer cancel()
 
 		var logs []tailscale.NetworkFlowLog
 
@@ -271,6 +279,7 @@ func (ts *TailscaleService) GetNetworkLogs(start, end string) (interface{}, erro
 			return nil, fmt.Errorf("failed to fetch network logs from tailscale client: %w", err)
 		}
 
+		
 		return map[string]interface{}{
 			"logs": logs,
 		}, nil
@@ -283,7 +292,12 @@ func (ts *TailscaleService) GetNetworkLogs(start, end string) (interface{}, erro
 		endpoint += fmt.Sprintf("?start=%s&end=%s", url.QueryEscape(start), url.QueryEscape(end))
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	// Use much longer timeout for larger time ranges
+	timeoutDuration := 10 * time.Minute
+	if endTime.Sub(startTime) > 7*24*time.Hour {
+		timeoutDuration = 30 * time.Minute // Much longer timeout for 30+ day queries
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
 	defer cancel()
 
 	body, err := ts.makeRequest(ctx, endpoint)
@@ -296,7 +310,17 @@ func (ts *TailscaleService) GetNetworkLogs(start, end string) (interface{}, erro
 		return nil, fmt.Errorf("failed to unmarshal network logs response: %w", err)
 	}
 
-	return response, nil
+	// Ensure consistent response format
+	if responseMap, ok := response.(map[string]interface{}); ok {
+		if logs, exists := responseMap["logs"]; exists {
+			return map[string]interface{}{
+				"logs": logs,
+			}, nil
+		}
+	}
+	return map[string]interface{}{
+		"logs": response,
+	}, nil
 }
 
 // GetNetworkLogsChunked retrieves network logs in chunks for large time ranges
@@ -660,3 +684,12 @@ func (ts *TailscaleService) GetStaticRecords() (map[string]StaticRecordInfo, err
 
 	return response.Records, nil
 }
+
+// Helper function to get map keys for debugging
+func getMapKeys(m map[string]interface{}) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys
+}

frontend/src/components/LogViewer.tsx

@@ -112,35 +112,41 @@ const processLogsWorker = (networkLogs: NetworkFlowLog[], devices: Device[]) =>
           const log = logs[i]
           const processTraffic = (traffic: TrafficFlow[], type: 'virtual' | 'subnet' | 'physical') => {
             traffic.forEach((flow) => {
-              const srcIP = extractIP(flow.src)
-              const dstIP = extractIP(flow.dst)
-              const timestamp = new Date(log.logged)
+              // Handle both capitalized and lowercase field names
+              const srcIP = extractIP((flow as any).Src || (flow as any).src)
+              const dstIP = extractIP((flow as any).Dst || (flow as any).dst)
+              const timestamp = new Date((log as any).Logged || (log as any).logged)
 
               entries.push({
-                id: `${log.nodeId}-${entryId++}`,
-                timestamp: log.logged,
+                id: `${(log as any).NodeID || (log as any).nodeId}-${entryId++}`,
+                timestamp: (log as any).Logged || (log as any).logged,
                 timestampMs: timestamp.getTime(),
-                nodeId: log.nodeId,
+                nodeId: (log as any).NodeID || (log as any).nodeId,
                 srcDevice: getDeviceName(srcIP, devices),
                 dstDevice: getDeviceName(dstIP, devices),
                 srcIP,
                 dstIP,
-                srcPort: extractPort(flow.src) || undefined,
-                dstPort: extractPort(flow.dst) || undefined,
-                protocol: getProtocolName(flow.proto),
+                srcPort: extractPort((flow as any).Src || (flow as any).src) || undefined,
+                dstPort: extractPort((flow as any).Dst || (flow as any).dst) || undefined,
+                protocol: getProtocolName((flow as any).Proto || (flow as any).proto),
                 trafficType: type,
-                txBytes: flow.txBytes || 0,
-                rxBytes: flow.rxBytes || 0,
-                txPackets: flow.txPackets || 0,
-                rxPackets: flow.rxPackets || 0,
+                txBytes: (flow as any).TxBytes || (flow as any).txBytes || 0,
+                rxBytes: (flow as any).RxBytes || (flow as any).rxBytes || 0,
+                txPackets: (flow as any).TxPkts || (flow as any).txPackets || 0,
+                rxPackets: (flow as any).RxPkts || (flow as any).rxPackets || 0,
                 tags: []
               })
             })
           }
 
-          if (log.virtualTraffic) processTraffic(log.virtualTraffic, 'virtual')
-          if (log.subnetTraffic) processTraffic(log.subnetTraffic, 'subnet')
-          if (log.physicalTraffic) processTraffic(log.physicalTraffic, 'physical')
+          // Handle both capitalized and lowercase field names
+          const virtualTraffic = (log as any).VirtualTraffic || (log as any).virtualTraffic
+          const subnetTraffic = (log as any).SubnetTraffic || (log as any).subnetTraffic
+          const physicalTraffic = (log as any).PhysicalTraffic || (log as any).physicalTraffic
+
+          if (virtualTraffic) processTraffic(virtualTraffic, 'virtual')
+          if (subnetTraffic) processTraffic(subnetTraffic, 'subnet')
+          if (physicalTraffic) processTraffic(physicalTraffic, 'physical')
         }
 
         if (endIndex < logs.length) {

frontend/src/pages/NetworkView.tsx

@@ -354,7 +354,11 @@ const NetworkView: React.FC = () => {
     }
 
     // Validate that we have actual log entries
-    return logsArray.filter(log => log && typeof log === 'object' && 'logged' in log)
+    const validLogs = logsArray.filter(log => {
+      return log && typeof log === 'object' && ('logged' in log || 'Logged' in log)
+    })
+    
+    return validLogs
   }, [networkLogsData])
 
   // Set default date range to show most recent data (last 5 minutes)