v2/logging: add support for network flow logs API

Adds support for the network flow logs API endpoint at
/api/v2/tailnet/{tailnet}/logging/network. This endpoint returns
network traffic flow data including virtual, subnet, exit, and
physical traffic with packet/byte counts.

Updates #41

Signed-off-by: rajsinghtech <[email protected]>

Author: rajsinghtech

logging.go

@@ -5,7 +5,12 @@ package tailscale
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"io"
 	"net/http"
+	"net/url"
+	"time"
 )
 
 // LoggingResource provides access to https://tailscale.com/api#tag/logging.
@@ -147,3 +152,133 @@ func (lr *LoggingResource) ValidateAWSTrustPolicy(ctx context.Context, awsExtern
 	}
 	return lr.do(req, nil)
 }
+
+// NetworkFlowLog represents a network flow log entry from the Tailscale API.
+type NetworkFlowLog struct {
+	Logged          time.Time      `json:"logged"`                    // the time at which this log was captured by the server
+	NodeID          string         `json:"nodeId"`                    // the node ID for which the flow statistics apply
+	Start           time.Time      `json:"start"`                     // the start of the sample period (node's local clock)
+	End             time.Time      `json:"end"`                       // the end of the sample period (node's local clock)
+	VirtualTraffic  []TrafficStats `json:"virtualTraffic,omitempty"`  // traffic between Tailscale nodes
+	SubnetTraffic   []TrafficStats `json:"subnetTraffic,omitempty"`   // traffic involving subnet routes
+	ExitTraffic     []TrafficStats `json:"exitTraffic,omitempty"`     // traffic via exit nodes
+	PhysicalTraffic []TrafficStats `json:"physicalTraffic,omitempty"` // WireGuard transport-level statistics
+}
+
+// TrafficStats represents traffic flow statistics.
+// This type is used for all traffic types: virtual, subnet, exit, and physical.
+type TrafficStats struct {
+	Proto   int    `json:"proto,omitempty"`   // IP protocol number (e.g., 6 for TCP, 17 for UDP)
+	Src     string `json:"src,omitempty"`     // Source address and port
+	Dst     string `json:"dst,omitempty"`     // Destination address and port
+	TxPkts  uint64 `json:"txPkts,omitempty"`  // Transmitted packets
+	TxBytes uint64 `json:"txBytes,omitempty"` // Transmitted bytes
+	RxPkts  uint64 `json:"rxPkts,omitempty"`  // Received packets
+	RxBytes uint64 `json:"rxBytes,omitempty"` // Received bytes
+}
+
+// NetworkFlowLogsRequest represents query parameters for fetching network flow logs.
+type NetworkFlowLogsRequest struct {
+	// Start must be set to a non-zero time within the log retention period (last 30 days).
+	// The server may adjust times that are too old.
+	Start time.Time
+	// End must be set to a non-zero time after Start.
+	End time.Time
+}
+
+// NetworkFlowLogHandler is a callback function for processing individual network flow log entries.
+// It receives each log entry as it's parsed from the JSON stream.
+// Return an error to stop processing and bubble up the error.
+type NetworkFlowLogHandler func(log NetworkFlowLog) error
+
+// GetNetworkFlowLogs streams network flow logs for the tailnet, calling the provided
+// handler function for each log entry as it's parsed from the JSON response.
+// This approach is memory-efficient and handles large datasets without loading all logs into memory.
+//
+// Both start and end parameters are required by the server.
+// Times older than 30 days will be automatically adjusted by the server to the retention limit.
+func (lr *LoggingResource) GetNetworkFlowLogs(ctx context.Context, params NetworkFlowLogsRequest, handler NetworkFlowLogHandler) error {
+
+	u := lr.buildTailnetURL("logging", "network")
+	u.RawQuery = url.Values{
+		"start": {params.Start.Format(time.RFC3339)},
+		"end":   {params.End.Format(time.RFC3339)},
+	}.Encode()
+
+	req, err := lr.buildRequest(ctx, http.MethodGet, u)
+	if err != nil {
+		return err
+	}
+
+	return lr.streamNetworkFlowLogs(req, handler)
+}
+
+// checkDelim reads and verifies the next JSON delimiter from the decoder
+func checkDelim(dec *json.Decoder, want json.Delim, description string) error {
+	token, err := dec.Token()
+	if err != nil {
+		return fmt.Errorf("failed to read %s: %w", description, err)
+	}
+	if delim, ok := token.(json.Delim); !ok || delim != want {
+		return fmt.Errorf("expected %c for %s, got %v", want, description, token)
+	}
+	return nil
+}
+
+// streamNetworkFlowLogs performs the streaming JSON parsing of network flow logs
+func (lr *LoggingResource) streamNetworkFlowLogs(req *http.Request, handler NetworkFlowLogHandler) error {
+	lr.init()
+	resp, err := lr.HTTP.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(body))
+	}
+
+	decoder := json.NewDecoder(resp.Body)
+
+	if err := checkDelim(decoder, '{', "opening brace"); err != nil {
+		return err
+	}
+
+	token, err := decoder.Token()
+	if err != nil {
+		return fmt.Errorf("failed to read field name: %w", err)
+	}
+	if fieldName, ok := token.(string); !ok || fieldName != "logs" {
+		return fmt.Errorf("expected 'logs' field, got %v", token)
+	}
+
+	if err := checkDelim(decoder, '[', "logs array start"); err != nil {
+		return err
+	}
+
+	for decoder.More() {
+		if err := req.Context().Err(); err != nil {
+			return err
+		}
+
+		var log NetworkFlowLog
+		if err := decoder.Decode(&log); err != nil {
+			return fmt.Errorf("failed to decode log entry: %w", err)
+		}
+
+		if err := handler(log); err != nil {
+			return fmt.Errorf("handler error: %w", err)
+		}
+	}
+
+	if err := checkDelim(decoder, ']', "logs array end"); err != nil {
+		return err
+	}
+
+	if err := checkDelim(decoder, '}', "closing brace"); err != nil {
+		return err
+	}
+
+	return nil
+}

logging_test.go

@@ -6,8 +6,10 @@ package tailscale
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -129,3 +131,80 @@ func TestClient_ValidateAWSTrustPolicy(t *testing.T) {
 	assert.NoError(t, err)
 	assert.EqualValues(t, gotRequest, map[string]string{"roleArn": roleARN})
 }
+
+func TestClient_GetNetworkFlowLogs(t *testing.T) {
+	t.Parallel()
+
+	client, server := NewTestHarness(t)
+	server.ResponseCode = http.StatusOK
+
+	now := time.Now().UTC().Truncate(time.Second)
+	expectedLogs := []NetworkFlowLog{
+		{
+			Logged: now,
+			NodeID: "node1",
+			Start:  now.Add(-5 * time.Minute),
+			End:    now,
+			VirtualTraffic: []TrafficStats{
+				{Proto: 6, Src: "10.0.0.1:80", Dst: "10.0.0.2:1234", TxPkts: 10, TxBytes: 1000},
+			},
+		},
+		{
+			Logged: now.Add(1 * time.Second),
+			NodeID: "node2",
+			Start:  now.Add(-4 * time.Minute),
+			End:    now.Add(1 * time.Second),
+			PhysicalTraffic: []TrafficStats{
+				{Proto: 17, Src: "192.168.1.1:53", Dst: "8.8.8.8:53", RxPkts: 5, RxBytes: 500},
+			},
+		},
+	}
+
+	server.ResponseBody = map[string]any{"logs": expectedLogs}
+
+	params := NetworkFlowLogsRequest{
+		Start: now.Add(-1 * time.Hour),
+		End:   now,
+	}
+
+	var actualLogs []NetworkFlowLog
+	handler := func(log NetworkFlowLog) error {
+		actualLogs = append(actualLogs, log)
+		return nil
+	}
+
+	err := client.Logging().GetNetworkFlowLogs(context.Background(), params, handler)
+	assert.NoError(t, err)
+	assert.Equal(t, http.MethodGet, server.Method)
+	assert.Equal(t, "/api/v2/tailnet/example.com/logging/network", server.Path)
+	
+	assert.Len(t, actualLogs, 2)
+	assert.Equal(t, expectedLogs, actualLogs)
+}
+
+func TestClient_GetNetworkFlowLogs_HandlerError(t *testing.T) {
+	t.Parallel()
+
+	client, server := NewTestHarness(t)
+	server.ResponseCode = http.StatusOK
+
+	now := time.Now().UTC()
+	server.ResponseBody = map[string]any{
+		"logs": []NetworkFlowLog{{
+			Logged: now, NodeID: "test-node", Start: now.Add(-5 * time.Minute), End: now,
+		}},
+	}
+
+	params := NetworkFlowLogsRequest{Start: now.Add(-1 * time.Hour), End: now}
+
+	handler := func(log NetworkFlowLog) error {
+		return fmt.Errorf("test handler error")
+	}
+
+	err := client.Logging().GetNetworkFlowLogs(context.Background(), params, handler)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "handler error")
+	assert.Contains(t, err.Error(), "test handler error")
+}
+
+