[proxy] refactor metrics and add usage logs

management/internals/modules/reverseproxy/accesslogs/accesslogentry.go

@@ -24,6 +24,8 @@ type AccessLogEntry struct {
 	Reason         string
 	UserId         string `gorm:"index"`
 	AuthMethodUsed string `gorm:"index"`
+	BytesUpload    int64  `gorm:"index"`
+	BytesDownload  int64  `gorm:"index"`
 }
 
 // FromProto creates an AccessLogEntry from a proto.AccessLog
@@ -39,6 +41,8 @@ func (a *AccessLogEntry) FromProto(serviceLog *proto.AccessLog) {
 	a.UserId = serviceLog.GetUserId()
 	a.AuthMethodUsed = serviceLog.GetAuthMechanism()
 	a.AccountID = serviceLog.GetAccountId()
+	a.BytesUpload = serviceLog.GetBytesUpload()
+	a.BytesDownload = serviceLog.GetBytesDownload()
 
 	if sourceIP := serviceLog.GetSourceIp(); sourceIP != "" {
 		if ip, err := netip.ParseAddr(sourceIP); err == nil {
@@ -101,5 +105,7 @@ func (a *AccessLogEntry) ToAPIResponse() *api.ProxyAccessLog {
 		AuthMethodUsed: authMethod,
 		CountryCode:    countryCode,
 		CityName:       cityName,
+		BytesUpload:    a.BytesUpload,
+		BytesDownload:  a.BytesDownload,
 	}
 }

proxy/internal/accesslog/logger.go

@@ -3,6 +3,7 @@ package accesslog
 import (
 	"context"
 	"net/netip"
+	"sync"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -13,6 +14,23 @@ import (
 	"github.com/netbirdio/netbird/shared/management/proto"
 )
 
+const (
+	requestThreshold    = 10000              // Log every 10k requests
+	bytesThreshold      = 1024 * 1024 * 1024 // Log every 1GB
+	usageCleanupPeriod  = 1 * time.Hour      // Clean up stale counters every hour
+	usageInactiveWindow = 24 * time.Hour     // Consider domain inactive if no traffic for 24 hours
+)
+
+type domainUsage struct {
+	requestCount     int64
+	requestStartTime time.Time
+
+	bytesTransferred int64
+	bytesStartTime   time.Time
+
+	lastActivity time.Time // Track last activity for cleanup
+}
+
 type gRPCClient interface {
 	SendAccessLog(ctx context.Context, in *proto.SendAccessLogRequest, opts ...grpc.CallOption) (*proto.SendAccessLogResponse, error)
 }
@@ -22,6 +40,11 @@ type Logger struct {
 	client         gRPCClient
 	logger         *log.Logger
 	trustedProxies []netip.Prefix
+
+	usageMux    sync.Mutex
+	domainUsage map[string]*domainUsage
+
+	cleanupCancel context.CancelFunc
 }
 
 // NewLogger creates a new access log Logger. The trustedProxies parameter
@@ -31,10 +54,26 @@ func NewLogger(client gRPCClient, logger *log.Logger, trustedProxies []netip.Pre
 	if logger == nil {
 		logger = log.StandardLogger()
 	}
-	return &Logger{
+
+	ctx, cancel := context.WithCancel(context.Background())
+	l := &Logger{
 		client:         client,
 		logger:         logger,
 		trustedProxies: trustedProxies,
+		domainUsage:    make(map[string]*domainUsage),
+		cleanupCancel:  cancel,
+	}
+
+	// Start background cleanup routine
+	go l.cleanupStaleUsage(ctx)
+
+	return l
+}
+
+// Close stops the cleanup routine. Should be called during graceful shutdown.
+func (l *Logger) Close() {
+	if l.cleanupCancel != nil {
+		l.cleanupCancel()
 	}
 }
 
@@ -51,6 +90,8 @@ type logEntry struct {
 	AuthMechanism string
 	UserId        string
 	AuthSuccess   bool
+	BytesUpload   int64
+	BytesDownload int64
 }
 
 func (l *Logger) log(ctx context.Context, entry logEntry) {
@@ -84,6 +125,8 @@ func (l *Logger) log(ctx context.Context, entry logEntry) {
 				AuthMechanism: entry.AuthMechanism,
 				UserId:        entry.UserId,
 				AuthSuccess:   entry.AuthSuccess,
+				BytesUpload:   entry.BytesUpload,
+				BytesDownload: entry.BytesDownload,
 			},
 		}); err != nil {
 			// If it fails to send on the gRPC connection, then at least log it to the error log.
@@ -103,3 +146,82 @@ func (l *Logger) log(ctx context.Context, entry logEntry) {
 		}
 	}()
 }
+
+// trackUsage records request and byte counts per domain, logging when thresholds are hit.
+func (l *Logger) trackUsage(domain string, bytesTransferred int64) {
+	if domain == "" {
+		return
+	}
+
+	l.usageMux.Lock()
+	defer l.usageMux.Unlock()
+
+	now := time.Now()
+	usage, exists := l.domainUsage[domain]
+	if !exists {
+		usage = &domainUsage{
+			requestStartTime: now,
+			bytesStartTime:   now,
+			lastActivity:     now,
+		}
+		l.domainUsage[domain] = usage
+	}
+
+	usage.lastActivity = now
+
+	usage.requestCount++
+	if usage.requestCount >= requestThreshold {
+		elapsed := time.Since(usage.requestStartTime)
+		l.logger.WithFields(log.Fields{
+			"domain":   domain,
+			"requests": usage.requestCount,
+			"duration": elapsed.String(),
+		}).Infof("domain %s had %d requests over %s", domain, usage.requestCount, elapsed)
+
+		usage.requestCount = 0
+		usage.requestStartTime = now
+	}
+
+	usage.bytesTransferred += bytesTransferred
+	if usage.bytesTransferred >= bytesThreshold {
+		elapsed := time.Since(usage.bytesStartTime)
+		bytesInGB := float64(usage.bytesTransferred) / (1024 * 1024 * 1024)
+		l.logger.WithFields(log.Fields{
+			"domain":   domain,
+			"bytes":    usage.bytesTransferred,
+			"bytes_gb": bytesInGB,
+			"duration": elapsed.String(),
+		}).Infof("domain %s transferred %.2f GB over %s", domain, bytesInGB, elapsed)
+
+		usage.bytesTransferred = 0
+		usage.bytesStartTime = now
+	}
+}
+
+// cleanupStaleUsage removes usage entries for domains that have been inactive.
+func (l *Logger) cleanupStaleUsage(ctx context.Context) {
+	ticker := time.NewTicker(usageCleanupPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			l.usageMux.Lock()
+			now := time.Now()
+			removed := 0
+			for domain, usage := range l.domainUsage {
+				if now.Sub(usage.lastActivity) > usageInactiveWindow {
+					delete(l.domainUsage, domain)
+					removed++
+				}
+			}
+			l.usageMux.Unlock()
+
+			if removed > 0 {
+				l.logger.Debugf("cleaned up %d stale domain usage entries", removed)
+			}
+		}
+	}
+}

proxy/internal/accesslog/middleware.go

@@ -32,6 +32,14 @@ func (l *Logger) Middleware(next http.Handler) http.Handler {
 			status:            http.StatusOK,
 		}
 
+		var bytesRead int64
+		if r.Body != nil {
+			r.Body = &bodyCounter{
+				ReadCloser: r.Body,
+				bytesRead:  &bytesRead,
+			}
+		}
+
 		// Resolve the source IP using trusted proxy configuration before passing
 		// the request on, as the proxy will modify forwarding headers.
 		sourceIp := extractSourceIP(r, l.trustedProxies)
@@ -53,6 +61,9 @@ func (l *Logger) Middleware(next http.Handler) http.Handler {
 			host = r.Host
 		}
 
+		bytesUpload := bytesRead
+		bytesDownload := sw.bytesWritten
+
 		entry := logEntry{
 			ID:            requestID,
 			ServiceId:     capturedData.GetServiceId(),
@@ -66,10 +77,15 @@ func (l *Logger) Middleware(next http.Handler) http.Handler {
 			AuthMechanism: capturedData.GetAuthMethod(),
 			UserId:        capturedData.GetUserID(),
 			AuthSuccess:   sw.status != http.StatusUnauthorized && sw.status != http.StatusForbidden,
+			BytesUpload:   bytesUpload,
+			BytesDownload: bytesDownload,
 		}
 		l.logger.Debugf("response: request_id=%s method=%s host=%s path=%s status=%d duration=%dms source=%s origin=%s service=%s account=%s",
 			requestID, r.Method, host, r.URL.Path, sw.status, duration.Milliseconds(), sourceIp, capturedData.GetOrigin(), capturedData.GetServiceId(), capturedData.GetAccountId())
 
 		l.log(r.Context(), entry)
+
+		// Track usage for cost monitoring (upload + download) by domain
+		l.trackUsage(host, bytesUpload+bytesDownload)
 	})
 }

proxy/internal/accesslog/statuswriter.go

@@ -1,18 +1,39 @@
 package accesslog
 
 import (
+	"io"
+
 	"github.com/netbirdio/netbird/proxy/internal/responsewriter"
 )
 
-// statusWriter captures the HTTP status code from WriteHeader calls.
+// statusWriter captures the HTTP status code and bytes written from responses.
 // It embeds responsewriter.PassthroughWriter which handles all the optional
 // interfaces (Hijacker, Flusher, Pusher) automatically.
 type statusWriter struct {
 	*responsewriter.PassthroughWriter
-	status int
+	status       int
+	bytesWritten int64
 }
 
 func (w *statusWriter) WriteHeader(status int) {
 	w.status = status
 	w.PassthroughWriter.WriteHeader(status)
 }
+
+func (w *statusWriter) Write(b []byte) (int, error) {
+	n, err := w.PassthroughWriter.Write(b)
+	w.bytesWritten += int64(n)
+	return n, err
+}
+
+// bodyCounter wraps an io.ReadCloser and counts bytes read from the request body.
+type bodyCounter struct {
+	io.ReadCloser
+	bytesRead *int64
+}
+
+func (bc *bodyCounter) Read(p []byte) (int, error) {
+	n, err := bc.ReadCloser.Read(p)
+	*bc.bytesRead += int64(n)
+	return n, err
+}

proxy/internal/acme/manager.go

@@ -42,6 +42,10 @@
 	err       string
 }
 
+type metricsRecorder interface {
+	RecordCertificateIssuance(duration time.Duration)
+}
+
 // Manager wraps autocert.Manager with domain tracking and cross-replica
 // coordination via a pluggable locking strategy. The locker prevents
 // duplicate ACME requests when multiple replicas share a certificate cache.
@@ -55,6 +59,7 @@
 
 	certNotifier certificateNotifier
 	logger       *log.Logger
+	metrics      metricsRecorder
 }
 
 // NewManager creates a new ACME certificate manager. The certDir is used
@@ -63,7 +68,7 @@
 // eabKID and eabHMACKey are optional External Account Binding credentials
 // required for some CAs like ZeroSSL. The eabHMACKey should be the base64
 // URL-encoded string provided by the CA.
-func NewManager(certDir, acmeURL, eabKID, eabHMACKey string, notifier certificateNotifier, logger *log.Logger, lockMethod CertLockMethod) *Manager {
+func NewManager(certDir, acmeURL, eabKID, eabHMACKey string, notifier certificateNotifier, logger *log.Logger, lockMethod CertLockMethod, metrics metricsRecorder) *Manager {
 	if logger == nil {
 		logger = log.StandardLogger()
 	}
@@ -73,6 +78,7 @@
 		domains:      make(map[domain.Domain]*domainInfo),
 		certNotifier: notifier,
 		logger:       logger,
+		metrics:      metrics,
 	}
 
 	var eab *acme.ExternalAccountBinding
@@ -161,6 +167,10 @@
 		return
 	}
 
+	if mgr.metrics != nil {
+		mgr.metrics.RecordCertificateIssuance(elapsed)
+	}
+
 	mgr.setDomainState(d, domainReady, "")
 
 	now := time.Now()

proxy/internal/acme/manager_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestHostPolicy(t *testing.T) {
-	mgr := NewManager(t.TempDir(), "https://acme.example.com/directory", "", "", nil, nil, "")
+	mgr := NewManager(t.TempDir(), "https://acme.example.com/directory", "", "", nil, nil, "", nil)
 	mgr.AddDomain("example.com", "acc1", "rp1")
 
 	// Wait for the background prefetch goroutine to finish so the temp dir
@@ -70,7 +70,7 @@ func TestHostPolicy(t *testing.T) {
 }
 
 func TestDomainStates(t *testing.T) {
-	mgr := NewManager(t.TempDir(), "https://acme.example.com/directory", "", "", nil, nil, "")
+	mgr := NewManager(t.TempDir(), "https://acme.example.com/directory", "", "", nil, nil, "", nil)
 
 	assert.Equal(t, 0, mgr.PendingCerts(), "initially zero")
 	assert.Equal(t, 0, mgr.TotalDomains(), "initially zero domains")