diff --git a/.gitignore b/.gitignore index c294e4dc..3a1ccd03 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,7 @@ # Build build/ +bin/ log/ vendor/ diff --git a/README.md b/README.md index bcbc5f59..df28800d 100644 --- a/README.md +++ b/README.md @@ -11,4 +11,4 @@ |v3.4.0| v1.2.1 | >= 0.8.6 and < 0.9.0 | |v3.3.0| v1.2.0 | >= 0.8.1 and < 0.9.0 | |v3.2.1| v1.1.0 | >= 0.7.0 and < 0.7.0 | -|v3.2.0| v1.1.0 | >= 0.7.0 and < 0.7.0 | +|v3.2.0| v1.1.0 | >= 0.7.0 and < 0.7.0 | \ No newline at end of file diff --git a/go.mod b/go.mod index d32b0a6c..f9e3d0f6 100644 --- a/go.mod +++ b/go.mod @@ -17,6 +17,7 @@ require ( github.com/stretchr/testify v1.10.0 github.com/urfave/cli/v2 v2.27.7 github.com/wmnsk/go-pfcp v0.0.23-0.20231009074152-d5a9c1f47114 + gopkg.in/h2non/gock.v1 v1.1.2 gopkg.in/yaml.v2 v2.4.0 ) @@ -35,6 +36,7 @@ require ( github.com/goccy/go-json v0.10.2 // indirect github.com/golang-jwt/jwt/v5 v5.2.2 // indirect github.com/google/go-cmp v0.6.0 // indirect + github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/leodido/go-urn v1.4.0 // indirect diff --git a/go.sum b/go.sum index 58b514e9..dc66d1ea 100644 --- a/go.sum +++ b/go.sum @@ -38,6 +38,8 @@ github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVI github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 h1:2VTzZjLZBgl62/EtslCrtky5vbi9dd7HrQPQIx6wqiw= +github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542/go.mod h1:Ow0tF8D4Kplbc8s8sSb3V2oUCygFHVp8gC3Dn6U4MNI= github.com/hashicorp/go-version v1.6.0 h1:feTTfFNnjP967rlCxM/I9g701jU+RN74YKx2mOkIeek= github.com/hashicorp/go-version v1.6.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= @@ -64,6 +66,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32 h1:W6apQkHrMkS0Muv8G/TipAy/FJl/rCYT0+EuS8+Z0z4= +github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32/go.mod h1:9wM+0iRr9ahx58uYLpLIr5fm8diHn0JbqRycJi6w0Ms= github.com/pascaldekloe/goe v0.1.1 h1:Ah6WQ56rZONR3RW3qWa2NCZ6JAVvSpUcoLBaOmYFt9Q= github.com/pascaldekloe/goe v0.1.1/go.mod h1:KSyfaxQOh0HZPjDP1FL/kFtbqYqrALJTaMafFUIccqU= github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= @@ -119,6 +123,8 @@ google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/g google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/h2non/gock.v1 v1.1.2 h1:jBbHXgGBK/AoPVfJh5x4r/WxIrElvbLel8TCZkkZJoY= +gopkg.in/h2non/gock.v1 v1.1.2/go.mod h1:n7UGz/ckNChHiK05rDoiC4MYSunEC/lyaUm2WWaDva0= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/ees/aggregator.go b/internal/ees/aggregator.go new file mode 100644 index 00000000..0cc2d2ff --- /dev/null +++ b/internal/ees/aggregator.go @@ -0,0 +1,489 @@ +// Package ees - UPF Event Exposure Service (EES) +// aggregator.go: periodic/on-demand reporting loop for USER_DATA_USAGE_MEASURES. +// +// MVP behavior (interval semantics): +// - Source.SnapshotNow() returns per-session counters for the provider's current interval +// (UL/DL bytes/packets with StartTime/EndTime). +// - Aggregator uses the current interval values as-is (no subtraction from previous snapshots). +// - Throughput is derived as (bytes * 8) / (EndTime - StartTime) in seconds, with guards. +// - "Clean up unused source": after each tick, for every subscription, remove entries in +// subscription.Snapshots whose SessionKey is absent in the current snapshot keys. +// - ModeOnDemand: deliver one immediate report using the current interval data, then refresh +// snapshots and switch the subscription.Mode to PERIODIC. + +package ees + +import ( + "context" + "sync" + "time" + + "github.com/sirupsen/logrus" + + "github.com/free5gc/go-upf/internal/report" +) + +// Aggregator accumulates usage reports pushed from the kernel and periodically +// sends notifications to subscribers. This is a pure Push model - no active polling. +// PerioServerInterface provides access to URR period information +type PerioServerInterface interface { + GetAnyURRPeriod(urrid uint32) time.Duration +} + +type Aggregator struct { + subscriptionStore *SubscriptionStore + reportPeriod time.Duration + notifier *Notifier + logger *logrus.Entry + + // [Push Mode] Mutex protects reportBuffer + mu sync.Mutex + // [Push Mode] Consolidated reports per subscription: Key=SubscriptionID, SubKey=SessionKey + reportBuffer map[string]map[SessionKey]*UsageMeasures + + // [Push Mode] Mutex protects TickOnce from concurrent execution + tickMu sync.Mutex + + sessionProvider SessionProvider // Added: to lookup UE IP + perioServer PerioServerInterface // Added: to query URR periods + + // Dynamic period adjustment + ticker *time.Ticker + tickerMu sync.Mutex + tickerReset chan struct{} // Signal to reset ticker when period changes + periodAdjusted bool // Marks if period has been adjusted once +} + +// NewAggregator constructs an Aggregator for pure Push mode. +// Reports are accumulated via PushReport() and sent periodically by TickOnce(). +func NewAggregator( + subscriptionStore *SubscriptionStore, + reportPeriod time.Duration, + notifier *Notifier, + logger *logrus.Entry, + sessionProvider SessionProvider, + perioServer PerioServerInterface, +) *Aggregator { + return &Aggregator{ + subscriptionStore: subscriptionStore, + reportPeriod: reportPeriod, + notifier: notifier, + logger: logger, + sessionProvider: sessionProvider, + perioServer: perioServer, + + // Initialize report buffer for Push mode + reportBuffer: make(map[string]map[SessionKey]*UsageMeasures), + + // Initialize ticker reset channel + tickerReset: make(chan struct{}, 1), + } +} + +// Run starts the periodic loop until ctx is done. +// Uses a dual-loop structure to handle ticker replacement when period is adjusted. +func (aggregator *Aggregator) Run(parentContext context.Context) { + aggregator.logger.Infof("ees aggregator started (reportPeriod: %v)", aggregator.reportPeriod) + + // Outer loop: recreate ticker when reset signal is received + for { + // Create/recreate ticker with current period + aggregator.tickerMu.Lock() + if aggregator.ticker != nil { + aggregator.ticker.Stop() + } + aggregator.ticker = time.NewTicker(aggregator.reportPeriod) + tickerC := aggregator.ticker.C // Capture channel before unlocking + currentPeriod := aggregator.reportPeriod + aggregator.tickerMu.Unlock() + + aggregator.logger.Debugf("ees ticker initialized (period: %v)", currentPeriod) + + // Inner loop: process ticks until reset or shutdown + tickerLoop: + for { + select { + case <-parentContext.Done(): + aggregator.tickerMu.Lock() + if aggregator.ticker != nil { + aggregator.ticker.Stop() + } + aggregator.tickerMu.Unlock() + aggregator.logger.Info("ees aggregator stopped") + return + case <-aggregator.tickerReset: + // Ticker needs to be recreated with new period + aggregator.logger.Info("ees ticker reset signal received, recreating ticker") + break tickerLoop + case <-tickerC: + if _, err := aggregator.TickOnce(parentContext); err != nil { + aggregator.logger.Warnf("ees aggregator tick failed: %v", err) + } + } + } + } +} + +// TickOnce sends notifications using accumulated reports from the Push buffer. +// This is the Pure Push model - no active polling. +// Reports for the same session are consolidated into a single report. +// Each subscription's reportPeriod is respected - notifications are only sent +// when enough time has passed since the last notification. +// Returns number of notifications attempted (sum over all subscriptions) and any error. +func (aggregator *Aggregator) TickOnce(ctx context.Context) (int, error) { + _ = ctx // ctx reserved for future cancellation support + + // Prevent concurrent TickOnce executions + aggregator.tickMu.Lock() + defer aggregator.tickMu.Unlock() + + now := time.Now() + totalNotifications := 0 + + // Lock and swap out the buffer to process current batch + aggregator.mu.Lock() + bufferedReports := aggregator.reportBuffer + aggregator.reportBuffer = make(map[string]map[SessionKey]*UsageMeasures) + aggregator.mu.Unlock() + + aggregator.logger.Infof("ees tick - processing buffer (subscriptionCount: %d, currentTime: %v)", + len(bufferedReports), now) + + // Iterate over all subscriptions + subscriptions := aggregator.subscriptionStore.AllSubscriptions() + + for _, subscription := range subscriptions { + // Use subscription mutex for field access + subscription.mu.Lock() + + // MVP scope: only USER_DATA_USAGE_MEASURES + perPduSession + if subscription.Event != EventUserDataUsageMeasures || + subscription.Granularity != GranularityPerSession { + subscription.mu.Unlock() + continue + } + + // Get consolidated reports for this subscription + sessionMap, hasReports := bufferedReports[subscription.ID] + + // [Special Case] For ONE_TIME (OnDemand) mode, if there is no data in buffer, + // we MUST send a zero-value report to fulfill the "immediate" requirement + // and then delete the subscription. + if subscription.Mode == ModeOnDemand && (!hasReports || len(sessionMap) == 0) { + aggregator.logger.Infof("ees ONE_TIME subscription %s has no buffered data, "+ + "sending zero-value report", subscription.ID) + + // Try to find active sessions for this UE to provide at least some identity info + zeroReports := aggregator.generateZeroReports(subscription) + + // We can unlock while calling Notifier as it only uses immutable fields or local variables + subscription.mu.Unlock() + + if len(zeroReports) > 0 { + if err := aggregator.notifier.Notify(subscription, zeroReports); err != nil { + aggregator.logger.Warnf("ees failed to send zero-value ONE_TIME report: %v", err) + // Keep in store for retry next tick (will be zero again if no data) + continue + } + totalNotifications++ + } else { + aggregator.logger.Warnf("ees ONE_TIME sub %s: no active sessions, skipping notification", + subscription.ID) + } + + // Implicit delete after "attempting" to report (or if no sessions exist to report on) + if err := aggregator.subscriptionStore.DeleteSubscription(subscription.ID); err != nil { + aggregator.logger.Warnf("ees failed to delete ONE_TIME sub %s: %v", subscription.ID, err) + } + aggregator.logger.Infof("ees ONE_TIME subscription %s implicitly deleted (empty/zero path)", subscription.ID) + continue + } + if !hasReports || len(sessionMap) == 0 { + subscription.mu.Unlock() + continue + } + + // Check if enough time has passed since last notification + if subscription.Mode == ModePeriodic { + periodDuration := time.Duration(subscription.PeriodSec) * time.Second + timeSinceLastNotify := now.Sub(subscription.LastNotify) + + if timeSinceLastNotify < (periodDuration - 100*time.Millisecond) { + // Keep reports by merging back into main buffer + aggregator.mu.Lock() + if _, ok := aggregator.reportBuffer[subscription.ID]; !ok { + aggregator.reportBuffer[subscription.ID] = make(map[SessionKey]*UsageMeasures) + } + for _, v := range sessionMap { + mergeMeasure(aggregator.reportBuffer[subscription.ID], v) + } + aggregator.mu.Unlock() + subscription.mu.Unlock() + continue + } + } + + // Prepare consolidated list for notification + consolidatedList := make([]UsageMeasures, 0, len(sessionMap)) + for _, m := range sessionMap { + computeThroughputIfPossible(m) + consolidatedList = append(consolidatedList, *m) + } + + // Unlock during network call + subscription.mu.Unlock() + + // Perform notification + if err := aggregator.notifier.Notify(subscription, consolidatedList); err != nil { + aggregator.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + "items": len(consolidatedList), + }).Warnf("ees notification failed (will retry next tick): %v", err) + + // Keep reports for retry in next tick + aggregator.mu.Lock() + if _, ok := aggregator.reportBuffer[subscription.ID]; !ok { + aggregator.reportBuffer[subscription.ID] = make(map[SessionKey]*UsageMeasures) + } + for _, v := range consolidatedList { + vCopy := v + mergeMeasure(aggregator.reportBuffer[subscription.ID], &vCopy) + } + aggregator.mu.Unlock() + } else { + aggregator.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + "items": len(consolidatedList), + }).Info("ees notification sent successfully") + + // Update last notify time + subscription.mu.Lock() + subscription.LastNotify = now + subscription.mu.Unlock() + + totalNotifications++ + + // 3GPP TS 29.564: ONE_TIME subscriptions are deleted implicitly after reporting. + if subscription.Mode == ModeOnDemand { + if err := aggregator.subscriptionStore.DeleteSubscription(subscription.ID); err != nil { + aggregator.logger.Warnf("ees failed to implicitly delete ONE_TIME subscription %s: %v", + subscription.ID, err) + } else { + aggregator.logger.Infof("ees ONE_TIME subscription %s implicitly deleted", subscription.ID) + } + } + } + } + + return totalNotifications, nil +} + +// mergeMeasure merges a new measure into an existing map of consolidated measures. +func mergeMeasure(m map[SessionKey]*UsageMeasures, newM *UsageMeasures) { + existing, ok := m[newM.Key] + if !ok { + m[newM.Key] = newM + return + } + + if newM.StartTime.Before(existing.StartTime) { + existing.StartTime = newM.StartTime + } + if newM.EndTime.After(existing.EndTime) { + existing.EndTime = newM.EndTime + } + + existing.ULBytesDelta += newM.ULBytesDelta + existing.DLBytesDelta += newM.DLBytesDelta + existing.ULPacketsDelta += newM.ULPacketsDelta + existing.DLPacketsDelta += newM.DLPacketsDelta + + // UeIpv4Addr should be the same, but update if existing is empty + if existing.UeIpv4Addr == "" { + existing.UeIpv4Addr = newM.UeIpv4Addr + } +} + +// PushReport handles unsolicited reports (e.g. from Kernel via Handler). +func (aggregator *Aggregator) PushReport(sessRpt report.SessReport) { + aggregator.logger.Debugf("PushReport called (SEID: %d, ReportCount: %d)", + sessRpt.SEID, len(sessRpt.Reports)) + + var ueIpv4Addr string + if aggregator.sessionProvider != nil { + if ip, ok := aggregator.sessionProvider.GetSessionContextUEIP(sessRpt.SEID); ok { + ueIpv4Addr = ip + } + } + + var totalUL, totalDL, totalULPkt, totalDLPkt uint64 + var startTime, endTime time.Time + reportCount := 0 + + for _, r := range sessRpt.Reports { + if r.Type() != report.USAR { + continue + } + usarep, ok := r.(report.USAReport) + if !ok { + continue + } + + if usarep.URRID != 2 { + continue + } + + totalUL += usarep.VolumMeasure.UplinkVolume + totalDL += usarep.VolumMeasure.DownlinkVolume + totalULPkt += usarep.VolumMeasure.UplinkPktNum + totalDLPkt += usarep.VolumMeasure.DownlinkPktNum + + if reportCount == 0 || usarep.StartTime.Before(startTime) { + startTime = usarep.StartTime + } + if reportCount == 0 || usarep.EndTime.After(endTime) { + endTime = usarep.EndTime + } + + reportCount++ + } + + if reportCount == 0 { + return + } + + m := &UsageMeasures{ + Key: SessionKey{LocalSEID: sessRpt.SEID}, + ULBytesDelta: totalUL, + DLBytesDelta: totalDL, + ULPacketsDelta: totalULPkt, + DLPacketsDelta: totalDLPkt, + StartTime: startTime, + EndTime: endTime, + UeIpv4Addr: ueIpv4Addr, + } + + subscriptions := aggregator.subscriptionStore.AllSubscriptions() + aggregator.mu.Lock() + defer aggregator.mu.Unlock() + + for _, sub := range subscriptions { + if !aggregator.matchesSubscription(sub, ueIpv4Addr) { + continue + } + + if _, ok := aggregator.reportBuffer[sub.ID]; !ok { + aggregator.reportBuffer[sub.ID] = make(map[SessionKey]*UsageMeasures) + } + + // Create a copy for this subscription since mergeMeasure might modify it + mCopy := *m + mergeMeasure(aggregator.reportBuffer[sub.ID], &mCopy) + + aggregator.logger.WithFields(logrus.Fields{ + "subscriptionId": sub.ID, + "seid": sessRpt.SEID, + "ueIp": ueIpv4Addr, + "ulBytes": totalUL, + "dlBytes": totalDL, + "urrCount": reportCount, + }).Debug("ees smf urr report captured and consolidated") + } +} + +func (aggregator *Aggregator) generateZeroReports(sub *Subscription) []UsageMeasures { + var reports []UsageMeasures + if aggregator.sessionProvider == nil { + return reports + } + + sessions := aggregator.sessionProvider.GetSessionContexts() + now := time.Now() + + for lSeid, ctx := range sessions { + if aggregator.matchesSubscription(sub, ctx.UeIPv4Addr) { + reports = append(reports, UsageMeasures{ + Key: SessionKey{LocalSEID: lSeid, RemoteSEID: ctx.RemoteSEID}, + StartTime: now.Add(-1 * time.Second), // minimal interval for zero report + EndTime: now, + UeIpv4Addr: ctx.UeIPv4Addr, + // All counter fields default to 0 + }) + } + } + return reports +} + +func (aggregator *Aggregator) matchesSubscription(sub *Subscription, ueIpv4Addr string) bool { + if sub.Granularity != GranularityPerSession { + aggregator.logger.WithFields(logrus.Fields{ + "subscriptionId": sub.ID, + "granularity": string(sub.Granularity), + }).Warn("ees granularity not supported, skipping") + return false + } + + if sub.Target.AnyUE { + return true + } + if sub.Target.UeIPAddress != "" && sub.Target.UeIPAddress == ueIpv4Addr { + return true + } + return false +} + +func computeThroughputIfPossible(usage *UsageMeasures) { + durationSeconds := usage.EndTime.Sub(usage.StartTime).Seconds() + if durationSeconds <= 0 { + return + } + usage.ULThroughputBps = (float64(usage.ULBytesDelta) * 8.0) / durationSeconds + usage.DLThroughputBps = (float64(usage.DLBytesDelta) * 8.0) / durationSeconds + usage.ULPacketThroughputPps = float64(usage.ULPacketsDelta) / durationSeconds + usage.DLPacketThroughputPps = float64(usage.DLPacketsDelta) / durationSeconds +} + +func (aggregator *Aggregator) AdjustReportPeriod(urrPeriod time.Duration) bool { + aggregator.tickerMu.Lock() + defer aggregator.tickerMu.Unlock() + + if aggregator.periodAdjusted { + aggregator.logger.Debug("ees period already adjusted, skipping") + return false + } + + if urrPeriod <= 0 { + aggregator.logger.Warnf("ees invalid URR period for adjustment: %v", urrPeriod) + return false + } + + urrPeriodSec := int(urrPeriod.Seconds()) + currentPeriodSec := int(aggregator.reportPeriod.Seconds()) + + var newPeriodSec int + if currentPeriodSec < urrPeriodSec { + newPeriodSec = urrPeriodSec + } else if currentPeriodSec%urrPeriodSec != 0 { + newPeriodSec = ((currentPeriodSec / urrPeriodSec) + 1) * urrPeriodSec + } else { + aggregator.periodAdjusted = true + aggregator.logger.Infof("ees period already optimal (current: %ds, urr: %ds)", currentPeriodSec, urrPeriodSec) + return false + } + + newPeriod := time.Duration(newPeriodSec) * time.Second + aggregator.reportPeriod = newPeriod + aggregator.periodAdjusted = true + + aggregator.logger.Infof("ees aggregator period adjusted (old: %ds, new: %ds, urr: %ds)", + currentPeriodSec, newPeriodSec, urrPeriodSec) + + select { + case aggregator.tickerReset <- struct{}{}: + aggregator.logger.Debug("ees ticker reset signal sent") + default: + aggregator.logger.Debug("ees ticker reset signal already pending") + } + + return true +} diff --git a/internal/ees/api.go b/internal/ees/api.go new file mode 100644 index 00000000..74efe346 --- /dev/null +++ b/internal/ees/api.go @@ -0,0 +1,266 @@ +// Package ees - UPF Event Exposure Service (EES) +// api.go: minimal REST API for EES subscriptions (create/delete). +// +// Endpoints (MVP): +// POST /nupf-ee/v1/ee-subscriptions -> Create subscription +// DELETE /nupf-ee/v1/ee-subscriptions/{id} -> Delete subscription +// +// Scope & constraints (MVP): +// - Only USER_DATA_USAGE_MEASURES + perPduSession are accepted. +// - Mode supports PERIODIC and ON_DEMAND; ON_DEMAND triggers an immediate TickOnce(). +// - Uses SMF-provisioned URRs for data collection (no Shadow URR). + +package ees + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/sirupsen/logrus" +) + +// Server provides the minimal REST API for EES. +type Server struct { + subscriptionStore *SubscriptionStore + aggregator *Aggregator + logger *logrus.Entry + httpServer *http.Server +} + +// NewServer constructs a Server. +func NewServer( + store *SubscriptionStore, + aggregator *Aggregator, + logger *logrus.Entry, +) *Server { + return &Server{ + subscriptionStore: store, + aggregator: aggregator, + logger: logger, + } +} + +// Routes registers handlers into the given mux. +func (server *Server) Routes(mux *http.ServeMux) { + mux.HandleFunc("/nupf-ee/v1/ee-subscriptions", server.handleCreateSubscription) // POST + mux.HandleFunc("/nupf-ee/v1/ee-subscriptions/", server.handleDeleteSubscriptionByID) // DELETE /.../{id} +} + +// Serve starts an HTTP server on the given listen address. +func (server *Server) Serve(listenAddress string) error { + mux := http.NewServeMux() + server.Routes(mux) + + server.logger.Infof("ees api server listening at %s", listenAddress) + server.httpServer = &http.Server{ + Addr: listenAddress, + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + } + if err := server.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + return err + } + return nil +} + +// Shutdown gracefully stops the API server. +func (server *Server) Shutdown(ctx context.Context) error { + if server.httpServer != nil { + server.logger.Info("shutting down ees api server") + return server.httpServer.Shutdown(ctx) + } + return nil +} + +// ----- Request/Response models ----- + +type createSubscriptionRequest struct { + Subscription UpfEventSubscription `json:"subscription"` + // SupportedFeatures string `json:"supportedFeatures,omitempty"` // Not implemented yet +} + +type UpfEventSubscription struct { + NfID string `json:"nfId"` + EventList []UpfEvent `json:"eventList"` + EventNotifyURI string `json:"eventNotifyUri"` + NotifyCorrelationID string `json:"notifyCorrelationId"` + EventReportingMode UpfEventMode `json:"eventReportingMode"` + + // Targeting: choose one + UeIPAddress string `json:"ueIpAddress,omitempty"` + AnyUE bool `json:"anyUe,omitempty"` +} + +type UpfEvent struct { + Type string `json:"type"` + MeasurementTypes []string `json:"measurementTypes,omitempty"` + // TS 29.564: Required when type=USER_DATA_USAGE_MEASURES + GranularityOfMeasurement string `json:"granularityOfMeasurement,omitempty"` + // PER_SESSION, PER_APPLICATION, PER_FLOW + AppIds []string `json:"appIds,omitempty"` + // Required for PER_APPLICATION + TrafficFilters []FlowInformation `json:"trafficFilters,omitempty"` + // Required for PER_FLOW +} + +type UpfEventMode struct { + Trigger string `json:"trigger"` // "PERIODIC" | "ONE_TIME" + ReportPeriod int `json:"reportPeriod,omitempty"` // Seconds +} + +type createSubscriptionResponse struct { + Subscription UpfEventSubscription `json:"subscription"` + SubscriptionID string `json:"subscriptionId"` + // ReportList []NotificationItem `json:"reportList,omitempty"` // Not implemented in immediate response yet +} + +// ----- Handlers ----- + +// handleCreateSubscription handles POST /nupf-ee/v1/ee-subscriptions +func (server *Server) handleCreateSubscription(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + var inboundRequest createSubscriptionRequest + if err := json.NewDecoder(r.Body).Decode(&inboundRequest); err != nil { + http.Error(w, fmt.Sprintf("invalid json: %v", err), http.StatusBadRequest) + return + } + + subscriptionCandidate, validationErr := server.validateAndMapRequest(inboundRequest) + if validationErr != nil { + server.logger.Warnf("subscription validation failed: %v", validationErr) + http.Error(w, validationErr.Error(), http.StatusBadRequest) + return + } + + id, err := server.subscriptionStore.CreateSubscription(subscriptionCandidate) + if err != nil { + status := http.StatusInternalServerError + if errors.Is(err, ErrInvalidSubscription) { + status = http.StatusBadRequest + } + server.logger.Warnf("subscription store failed: %v", err) + http.Error(w, err.Error(), status) + return + } + server.logger.Infof("ees subscription created: id=%s nfId=%s trigger=%s", + id, inboundRequest.Subscription.NfID, inboundRequest.Subscription.EventReportingMode.Trigger) + + // If Mode is ON_DEMAND, trigger an immediate tick in the aggregator. + if subscriptionCandidate.Mode == ModeOnDemand && server.aggregator != nil { + server.logger.Infof("ees immediate tick triggered for subscriptionId=%s", id) + go func() { + if _, tickErr := server.aggregator.TickOnce(context.Background()); tickErr != nil { + server.logger.Warnf("ees on-demand immediate tick failed for subscriptionId=%s: %v", id, tickErr) + } + }() + } + + w.Header().Set("Location", "/nupf-ee/v1/ee-subscriptions/"+id) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + if encodeErr := json.NewEncoder(w).Encode(createSubscriptionResponse{ + Subscription: inboundRequest.Subscription, + SubscriptionID: id, + }); encodeErr != nil { + server.logger.Errorf("failed to encode response: %v", encodeErr) + } +} + +// handleDeleteSubscriptionByID handles DELETE /nupf-ee/v1/ee-subscriptions/{id} +func (server *Server) handleDeleteSubscriptionByID(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + path := strings.TrimSuffix(r.URL.Path, "/") + parts := strings.Split(path, "/") + if len(parts) < 4 { + http.Error(w, "not found", http.StatusNotFound) + return + } + id := parts[len(parts)-1] + if id == "" { + http.Error(w, "not found", http.StatusNotFound) + return + } + + if err := server.subscriptionStore.DeleteSubscription(id); err == nil { + server.logger.Infof("ees subscription deleted: id=%s", id) + w.WriteHeader(http.StatusNoContent) + } else { + server.logger.Warnf("ees subscription delete failed: not found: id=%s", id) + http.Error(w, "not found", http.StatusNotFound) + } +} + +// ----- Internal Helpers ----- + +func (server *Server) validateAndMapRequest(req createSubscriptionRequest) (*Subscription, error) { + sub := req.Subscription + if len(sub.EventList) == 0 { + return nil, errors.New("eventList must not be empty") + } + + // MVP: Only support the first event for now + e := sub.EventList[0] + if e.Type != string(EventUserDataUsageMeasures) { + return nil, fmt.Errorf("unsupported event type: %s", e.Type) + } + + // MVP: Only support PER_SESSION granularity + if e.GranularityOfMeasurement != string(GranularityPerSession) { + return nil, fmt.Errorf("unsupported granularity: %s (MVP only supports PER_SESSION)", e.GranularityOfMeasurement) + } + + var mode Mode + switch strings.ToUpper(sub.EventReportingMode.Trigger) { + case "PERIODIC": + mode = ModePeriodic + case "ONE_TIME": + mode = ModeOnDemand + default: + return nil, fmt.Errorf("unsupported trigger: %s", sub.EventReportingMode.Trigger) + } + + mTypes := make([]MeasurementType, 0, len(e.MeasurementTypes)) + for _, mt := range e.MeasurementTypes { + mtStr := MeasurementType(strings.ToUpper(mt)) + if mtStr != MeasureVolume && mtStr != MeasureThroughput && mtStr != MeasureAppInfo { + return nil, fmt.Errorf("unsupported measurementType: %s", mt) + } + mTypes = append(mTypes, mtStr) + } + + if sub.AnyUE && sub.UeIPAddress != "" { + return nil, errors.New("cannot specify both anyUe and ueIpAddress") + } + if !sub.AnyUE && sub.UeIPAddress == "" { + return nil, errors.New("must specify either anyUe or ueIpAddress") + } + + return &Subscription{ + Event: EventType(e.Type), + NotifURI: sub.EventNotifyURI, + NotifyCorrelationID: sub.NotifyCorrelationID, + NfID: sub.NfID, + Granularity: Granularity(e.GranularityOfMeasurement), + Mode: mode, + PeriodSec: sub.EventReportingMode.ReportPeriod, + MeasurementTypes: mTypes, + Target: TargetScope{ + UeIPAddress: sub.UeIPAddress, + AnyUE: sub.AnyUE, + }, + Snapshots: make(map[SessionKey]Counters), + }, nil +} diff --git a/internal/ees/api_test.go b/internal/ees/api_test.go new file mode 100644 index 00000000..f7abb7bb --- /dev/null +++ b/internal/ees/api_test.go @@ -0,0 +1,171 @@ +package ees + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestHandleCreateSubscription(t *testing.T) { + logger := logrus.NewEntry(logrus.New()) + store := NewSubscriptionStore("") + // Aggregator can be nil for basic API validation testing if we don't hit ModeOnDemand + server := NewServer(store, nil, logger) + + t.Run("ValidPeriodicSubscription", func(t *testing.T) { + reqBody := createSubscriptionRequest{ + Subscription: UpfEventSubscription{ + NfID: "test-nf", + EventNotifyURI: "http://localhost:9999/notify", + NotifyCorrelationID: "test-corr", + EventList: []UpfEvent{ + { + Type: "USER_DATA_USAGE_MEASURES", + GranularityOfMeasurement: "PER_SESSION", + MeasurementTypes: []string{"VOLUME_MEASUREMENT"}, + }, + }, + EventReportingMode: UpfEventMode{ + Trigger: "PERIODIC", + ReportPeriod: 10, + }, + AnyUE: true, + }, + } + + body, err := json.Marshal(reqBody) + require.NoError(t, err) + req := httptest.NewRequestWithContext( + context.Background(), + http.MethodPost, + "/nupf-ee/v1/ee-subscriptions", + bytes.NewReader(body), + ) + rr := httptest.NewRecorder() + + server.handleCreateSubscription(rr, req) + + assert.Equal(t, http.StatusCreated, rr.Code) + var resp createSubscriptionResponse + err = json.Unmarshal(rr.Body.Bytes(), &resp) + require.NoError(t, err) + assert.NotEmpty(t, resp.SubscriptionID) + + // Check if it was actually added to store + _, ok := store.GetSubscription(resp.SubscriptionID) + assert.True(t, ok) + }) + + t.Run("UnsupportedEventType", func(t *testing.T) { + reqBody := createSubscriptionRequest{ + Subscription: UpfEventSubscription{ + EventList: []UpfEvent{ + { + Type: "QOS_MONITORING", // Not supported in current MVP + }, + }, + }, + } + + body, err := json.Marshal(reqBody) + require.NoError(t, err) + req := httptest.NewRequestWithContext( + context.Background(), + http.MethodPost, + "/nupf-ee/v1/ee-subscriptions", + bytes.NewReader(body), + ) + rr := httptest.NewRecorder() + + server.handleCreateSubscription(rr, req) + + assert.Equal(t, http.StatusBadRequest, rr.Code) + assert.Contains(t, rr.Body.String(), "unsupported event type") + }) + + t.Run("MissingEventList", func(t *testing.T) { + reqBody := createSubscriptionRequest{ + Subscription: UpfEventSubscription{ + EventList: []UpfEvent{}, + }, + } + + body, err := json.Marshal(reqBody) + require.NoError(t, err) + req := httptest.NewRequestWithContext( + context.Background(), + http.MethodPost, + "/nupf-ee/v1/ee-subscriptions", + bytes.NewReader(body), + ) + rr := httptest.NewRecorder() + + server.handleCreateSubscription(rr, req) + + assert.Equal(t, http.StatusBadRequest, rr.Code) + assert.Contains(t, rr.Body.String(), "eventList must not be empty") + }) + + t.Run("UnsupportedGranularity", func(t *testing.T) { + reqBody := createSubscriptionRequest{ + Subscription: UpfEventSubscription{ + EventList: []UpfEvent{ + { + Type: "USER_DATA_USAGE_MEASURES", + GranularityOfMeasurement: "PER_APPLICATION", // Not supported in current MVP + }, + }, + }, + } + + body, err := json.Marshal(reqBody) + require.NoError(t, err) + req := httptest.NewRequestWithContext( + context.Background(), + http.MethodPost, + "/nupf-ee/v1/ee-subscriptions", + bytes.NewReader(body), + ) + rr := httptest.NewRecorder() + + server.handleCreateSubscription(rr, req) + + assert.Equal(t, http.StatusBadRequest, rr.Code) + assert.Contains(t, rr.Body.String(), "unsupported granularity") + }) + + t.Run("DeleteSubscription", func(t *testing.T) { + sub := &Subscription{ + NotifURI: "http://test", + Event: EventUserDataUsageMeasures, + Granularity: GranularityPerSession, + Mode: ModePeriodic, + PeriodSec: 10, + NfID: "test", + NotifyCorrelationID: "test", + } + id, err := store.CreateSubscription(sub) + require.NoError(t, err) + + req := httptest.NewRequestWithContext( + context.Background(), + http.MethodDelete, + "/nupf-ee/v1/ee-subscriptions/"+id, + nil, + ) + rr := httptest.NewRecorder() + + server.handleDeleteSubscriptionByID(rr, req) + + assert.Equal(t, http.StatusNoContent, rr.Code) + _, ok := store.GetSubscription(id) + assert.False(t, ok) + }) +} diff --git a/internal/ees/handler.go b/internal/ees/handler.go new file mode 100644 index 00000000..506c9175 --- /dev/null +++ b/internal/ees/handler.go @@ -0,0 +1,37 @@ +package ees + +import ( + "github.com/sirupsen/logrus" + + "github.com/free5gc/go-upf/internal/report" +) + +// Handler implements report.Handler for the EES module. +// It receives unsolicited usage reports from the kernel (via Dispatcher) +// and forwards them to the Aggregator (or processes them directly). +type Handler struct { + aggregator *Aggregator + logger *logrus.Entry +} + +func NewHandler(aggregator *Aggregator, logger *logrus.Entry) *Handler { + return &Handler{ + aggregator: aggregator, + logger: logger, + } +} + +// NotifySessReport is called when the Forwarder pushes a report (e.g., periodic URR). +func (h *Handler) NotifySessReport(sessRpt report.SessReport) { + h.logger.Infof("EES Handler received report from dispatcher (SEID: %d, ReportCount: %d)", + sessRpt.SEID, len(sessRpt.Reports)) + + // Filter: Check if these reports belong to EES (URR ID based or blind forwarding?) + // For MVP, we pass it to Aggregator.PushReport + h.aggregator.PushReport(sessRpt) +} + +// PopBufPkt is not supported by EES. +func (h *Handler) PopBufPkt(seid uint64, pdrid uint16) ([]byte, bool) { + return nil, false +} diff --git a/internal/ees/notifier.go b/internal/ees/notifier.go new file mode 100644 index 00000000..8ccb4924 --- /dev/null +++ b/internal/ees/notifier.go @@ -0,0 +1,219 @@ +// Package ees - UPF Event Exposure Service (EES) +// notifier.go: deliver EES Notify payloads to subscriber endpoints. +// +// Behavior (MVP): +// - Sends USER_DATA_USAGE_MEASURES notifications to sub.NotifURI via HTTP POST. +// - Payload contains subscription ID, event ID, granularity, and a list of items, +// each item includes LocalSEID, RemoteSEID, UL/DL bytes/packets, StartTime/EndTime, +// and optional derived throughputs. +// - Uses a per-request timeout; no retry (keep it simple for MVP). +// - Logs both success and failure with descriptive fields. + +package ees + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "time" + + "github.com/sirupsen/logrus" +) + +// Notifier is responsible for delivering EES notifications to subscribers. +type Notifier struct { + httpClient *http.Client + logger *logrus.Entry + defaultUserAgent string + requestTimeout time.Duration + maxResponseBodyLen int64 +} + +// NewNotifier creates a notifier with sane defaults. +// - request timeout: 5s +// - connect + TLS handshake timeouts are governed by the http.Transport below. +func NewNotifier(logger *logrus.Entry) *Notifier { + transport := &http.Transport{ + // Reasonable defaults for a control-plane style HTTP call. + DialContext: (&net.Dialer{ + Timeout: 3 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + MaxIdleConns: 64, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 3 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + } + return &Notifier{ + httpClient: &http.Client{ + Transport: transport, + }, + logger: logger, + defaultUserAgent: "go-upf-ees/notify", + requestTimeout: 5 * time.Second, + maxResponseBodyLen: 4 << 10, // 4 KiB cap when logging response bodies + } +} + +// NotificationData per TS 29.564 schema. +type NotificationData struct { + NotificationItems []NotificationItem `json:"notificationItems"` + CorrelationId string `json:"correlationId,omitempty"` +} + +// NotificationItem per TS 29.564 schema. +type NotificationItem struct { + // Required fields + EventType string `json:"eventType"` + TimeStamp time.Time `json:"timeStamp"` + + // Conditional: one of ueIpv4Addr, ueIpv6Prefix, or ueMacAddr must be present + UeIpv4Addr string `json:"ueIpv4Addr,omitempty"` + + // Optional fields + Supi string `json:"supi,omitempty"` + StartTime *time.Time `json:"startTime,omitempty"` + + // Measurements array per TS 29.564 + UserDataUsageMeasurements []UserDataUsageMeasurements `json:"userDataUsageMeasurements,omitempty"` +} + +// Notify converts a set of UsageMeasures into a POST request per TS 29.564. +func (notifier *Notifier) Notify(subscription *Subscription, measures []UsageMeasures) error { + if subscription == nil { + return fmt.Errorf("notify: subscription is nil") + } + if subscription.NotifURI == "" { + return fmt.Errorf("notify: empty NotifURI for subscriptionId=%s", subscription.ID) + } + + notificationItems := make([]NotificationItem, 0, len(measures)) + now := time.Now() + + for _, m := range measures { + item := NotificationItem{ + EventType: string(subscription.Event), + TimeStamp: now, + UeIpv4Addr: m.UeIpv4Addr, + StartTime: &m.StartTime, + } + + switch subscription.Event { + case EventUserDataUsageMeasures: + // Build UserDataUsageMeasurements per TS 29.564 + measurement := UserDataUsageMeasurements{} + + // Conditionally add Volume Measurement + if subscription.HasMeasurementType(MeasureVolume) { + measurement.VolumeMeasurement = &VolumeMeasurement{ + TotalVolume: m.ULBytesDelta + m.DLBytesDelta, + UlVolume: m.ULBytesDelta, + DlVolume: m.DLBytesDelta, + TotalNbOfPackets: m.ULPacketsDelta + m.DLPacketsDelta, + UlNbOfPackets: m.ULPacketsDelta, + DlNbOfPackets: m.DLPacketsDelta, + } + } + + // Conditionally add Throughput Measurement + if subscription.HasMeasurementType(MeasureThroughput) { + measurement.ThroughputMeasurement = &ThroughputMeasurement{ + UlThroughput: fmt.Sprintf("%.0f bps", m.ULThroughputBps), + DlThroughput: fmt.Sprintf("%.0f bps", m.DLThroughputBps), + UlPacketThroughput: fmt.Sprintf("%.2f pps", m.ULPacketThroughputPps), + DlPacketThroughput: fmt.Sprintf("%.2f pps", m.DLPacketThroughputPps), + } + } + + if measurement.VolumeMeasurement == nil && + measurement.ThroughputMeasurement == nil && + measurement.ThroughputStatisticsMeasurement == nil { + return fmt.Errorf("notify: subscriptionId=%s has no supported measurementTypes", subscription.ID) + } + item.UserDataUsageMeasurements = append(item.UserDataUsageMeasurements, measurement) + } + + notificationItems = append(notificationItems, item) + } + + payload := NotificationData{ + NotificationItems: notificationItems, + CorrelationId: subscription.NotifyCorrelationID, + } + + bodyBytes, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("notify: marshal payload failed: %w", err) + } + + // Context with timeout for the request lifecycle. + ctx, cancel := context.WithTimeout(context.Background(), notifier.requestTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, subscription.NotifURI, bytes.NewReader(bodyBytes)) + if err != nil { + return fmt.Errorf("notify: new request failed: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", notifier.defaultUserAgent) + req.Header.Set("X-EES-EventId", string(subscription.Event)) + req.Header.Set("X-EES-SubscriptionId", subscription.ID) + + resp, err := notifier.httpClient.Do(req) + if err != nil { + notifier.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + "notifUri": subscription.NotifURI, + "items": len(payload.NotificationItems), + }).Warnf("ees notify failed: http request error: %v", err) + return fmt.Errorf("notify: http request failed: %w", err) + } + + defer func() { + if closeErr := resp.Body.Close(); closeErr != nil { + notifier.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + }).Debugf("ees notify: close response body failed: %v", closeErr) + } + }() + + // Accept any 2xx as success. + if resp.StatusCode/100 != 2 { + // Best-effort small response read for logging; avoid large buffers. + var snippet string + limit := notifier.maxResponseBodyLen + if limit <= 0 { + limit = 4096 + } + bodyLimited := io.LimitReader(resp.Body, limit) + bodyBytes, readErr := io.ReadAll(bodyLimited) + if readErr != nil { + snippet = fmt.Sprintf("status=%s body=(read error: %v)", resp.Status, readErr) + } else if resp.ContentLength != -1 && resp.ContentLength > limit { + snippet = fmt.Sprintf("status=%s body=%q (truncated)", resp.Status, string(bodyBytes)) + } else { + snippet = fmt.Sprintf("status=%s body=%q", resp.Status, string(bodyBytes)) + } + + notifier.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + "notifUri": subscription.NotifURI, + "statusCode": resp.StatusCode, + "items": len(payload.NotificationItems), + }).Warnf("ees notify failed: non-2xx response: %s", snippet) + return fmt.Errorf("notify: non-2xx response: %s", resp.Status) + } + + // Success log (compact, with essential identifiers). + notifier.logger.WithFields(logrus.Fields{ + "subscriptionId": subscription.ID, + "notifUri": subscription.NotifURI, + "items": len(payload.NotificationItems), + }).Debug("ees notify success") + + return nil +} diff --git a/internal/ees/notifier_test.go b/internal/ees/notifier_test.go new file mode 100644 index 00000000..de292474 --- /dev/null +++ b/internal/ees/notifier_test.go @@ -0,0 +1,89 @@ +package ees + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "testing" + "time" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "gopkg.in/h2non/gock.v1" +) + +func TestNotifier_Notify(t *testing.T) { + defer gock.Off() + + logger := logrus.NewEntry(logrus.New()) + notifier := NewNotifier(logger) + gock.InterceptClient(notifier.httpClient) + + sub := &Subscription{ + ID: "sub-123", + Event: EventUserDataUsageMeasures, + NotifURI: "http://subscriber:8080/callback", + NotifyCorrelationID: "corr-123", + MeasurementTypes: []MeasurementType{MeasureVolume}, + } + + measures := []UsageMeasures{ + { + UeIpv4Addr: "10.0.0.1", + StartTime: time.Now().Add(-10 * time.Second), + EndTime: time.Now(), + ULBytesDelta: 1000, + DLBytesDelta: 2000, + ULPacketsDelta: 10, + DLPacketsDelta: 20, + }, + } + + t.Run("SuccessfulNotification", func(t *testing.T) { + gock.New("http://subscriber:8080"). + Post("/callback"). + Reply(200). + BodyString("OK") + + err := notifier.Notify(sub, measures) + assert.NoError(t, err) + assert.True(t, gock.IsDone()) + }) + + t.Run("CheckPayloadStructure", func(t *testing.T) { + gock.New("http://subscriber:8080"). + Post("/callback"). + MatchHeader("Content-Type", "application/json"). + // Check correlationId field in JSON body + Filter(func(req *http.Request) bool { + var data map[string]interface{} + body, err := io.ReadAll(req.Body) + if err != nil { + return false + } + req.Body = io.NopCloser(bytes.NewBuffer(body)) + if err := json.Unmarshal(body, &data); err != nil { + return false + } + return data["correlationId"] == "corr-123" + }). + Reply(204) + + err := notifier.Notify(sub, measures) + assert.NoError(t, err) + assert.True(t, gock.IsDone()) + }) + + t.Run("HandleHttpError", func(t *testing.T) { + gock.New("http://subscriber:8080"). + Post("/callback"). + Reply(500). + BodyString("Internal Server Error") + + err := notifier.Notify(sub, measures) + assert.Error(t, err) + assert.Contains(t, err.Error(), "non-2xx response") + assert.True(t, gock.IsDone()) + }) +} diff --git a/internal/ees/ondemand_test.go b/internal/ees/ondemand_test.go new file mode 100644 index 00000000..814b0991 --- /dev/null +++ b/internal/ees/ondemand_test.go @@ -0,0 +1,121 @@ +package ees + +import ( + "context" + "testing" + "time" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gopkg.in/h2non/gock.v1" +) + +func TestModeOnDemandLifecycle(t *testing.T) { + defer gock.Off() + logger := logrus.NewEntry(logrus.New()) + store := NewSubscriptionStore("") + notifier := NewNotifier(logger) + gock.InterceptClient(notifier.httpClient) + + // Mock Aggregator dependencies + aggregator := NewAggregator(store, 10*time.Second, notifier, logger, nil, nil) + + sub := &Subscription{ + NotifURI: "http://subscriber:8080/callback", + NotifyCorrelationID: "corr-ondemand", + Event: EventUserDataUsageMeasures, + Granularity: GranularityPerSession, + Mode: ModeOnDemand, + MeasurementTypes: []MeasurementType{MeasureVolume}, + NfID: "nf-ondemand", + } + id, err := store.CreateSubscription(sub) + require.NoError(t, err) + + t.Run("OnDemandWithData_ShouldNotifyAndDelete", func(t *testing.T) { + // 1. Manually inject data into aggregator buffer + aggregator.mu.Lock() + aggregator.reportBuffer[id] = map[SessionKey]*UsageMeasures{ + {LocalSEID: 1}: { + Key: SessionKey{LocalSEID: 1}, + ULBytesDelta: 100, + DLBytesDelta: 200, + StartTime: time.Now().Add(-5 * time.Second), + EndTime: time.Now(), + UeIpv4Addr: "10.0.0.1", + }, + } + aggregator.mu.Unlock() + + // 2. Mock the callback + gock.New("http://subscriber:8080"). + Post("/callback"). + Reply(204) + + // 3. Trigger TickOnce + n, err := aggregator.TickOnce(context.Background()) + assert.NoError(t, err) + assert.Equal(t, 1, n) + + // 4. Verify subscription is deleted + _, exists := store.GetSubscription(id) + assert.False(t, exists, "Subscription should be implicitly deleted after successful report") + assert.True(t, gock.IsDone()) + }) + + t.Run("OnDemandEmptyBuffer_ShouldSendZeroReportAndDelete", func(t *testing.T) { + // Mock SessionProvider to return an active session + mockProvider := &mockSessionProvider{ + sessions: map[uint64]SessionContext{ + 10: {UeIPv4Addr: "10.60.0.1", RemoteSEID: 99}, + }, + } + aggregator.sessionProvider = mockProvider + + // Create a new ONE_TIME subscription targeting the UE + sub2 := &Subscription{ + NotifURI: "http://subscriber:8080/callback-zero", + NotifyCorrelationID: "corr-zero", + Event: EventUserDataUsageMeasures, + Granularity: GranularityPerSession, + Mode: ModeOnDemand, + MeasurementTypes: []MeasurementType{MeasureVolume}, + NfID: "nf-ondemand", + Target: TargetScope{ + UeIPAddress: "10.60.0.1", + }, + } + id2, err := store.CreateSubscription(sub2) + require.NoError(t, err) + _ = id2 // id2 is used implicitly via store lookup in TickOnce + + // Mock the callback for zero report + gock.New("http://subscriber:8080"). + Post("/callback-zero"). + Reply(204) + + // Trigger TickOnce with empty buffer + n, err := aggregator.TickOnce(context.Background()) + assert.NoError(t, err) + assert.Equal(t, 1, n, "Should have sent 1 zero-value report") + + // Verify subscription is deleted + _, exists := store.GetSubscription(id2) + assert.False(t, exists, "Subscription should be implicitly deleted even with zero report") + assert.True(t, gock.IsDone()) + }) +} + +type mockSessionProvider struct { + sessions map[uint64]SessionContext +} + +func (m *mockSessionProvider) GetSessionContexts() map[uint64]SessionContext { + return m.sessions +} + +func (m *mockSessionProvider) GetSessionContextUEIP(lSeid uint64) (string, bool) { + ctx, ok := m.sessions[lSeid] + return ctx.UeIPv4Addr, ok +} diff --git a/internal/ees/subscription_store.go b/internal/ees/subscription_store.go new file mode 100644 index 00000000..a544e4cd --- /dev/null +++ b/internal/ees/subscription_store.go @@ -0,0 +1,149 @@ +// Package ees - UPF Event Exposure Service (EES) +// subscription_store.go: thread-safe in-memory store for EES subscriptions. +// +// Scope: +// - Only manages subscriptions (create/delete/get/list). +// - ID generation is local, simple, and collision-resistant enough for MVP. +package ees + +import ( + "errors" + "fmt" + "sync" + "time" +) + +// ErrSubscriptionNotFound is returned when a subscription ID does not exist. +var ErrSubscriptionNotFound = errors.New("subscription not found") + +// ErrInvalidSubscription is returned when a subscription is nil or missing required fields. +var ErrInvalidSubscription = errors.New("invalid subscription") + +// SubscriptionStore is a thread-safe in-memory storage for EES subscriptions. +// It intentionally uses descriptive field names for readability and maintenance. +type SubscriptionStore struct { + mutexForSubscriptions sync.RWMutex + subscriptionsByID map[string]*Subscription + + // simple in-process ID generator state + mutexForIDGen sync.Mutex + lastTimestampNS int64 + sequencePerTS uint32 + instanceSuffixStr string // optional, e.g., hostname or short random; empty is fine for MVP +} + +// NewSubscriptionStore creates a new, empty store for subscriptions. +func NewSubscriptionStore(instanceSuffix string) *SubscriptionStore { + return &SubscriptionStore{ + subscriptionsByID: make(map[string]*Subscription), + instanceSuffixStr: instanceSuffix, + } +} + +// CreateSubscription inserts a subscription and returns its generated ID. +// Minimal validation is performed to reduce accidental bad states. +func (store *SubscriptionStore) CreateSubscription(newSubscription *Subscription) (string, error) { + if newSubscription == nil { + return "", ErrInvalidSubscription + } + if newSubscription.NotifURI == "" { + return "", fmt.Errorf("%w: missing NotifURI", ErrInvalidSubscription) + } + if newSubscription.Event == "" { + return "", fmt.Errorf("%w: missing Event", ErrInvalidSubscription) + } + if newSubscription.Granularity == "" { + return "", fmt.Errorf("%w: missing Granularity", ErrInvalidSubscription) + } + if newSubscription.Mode == "" { + return "", fmt.Errorf("%w: missing Mode", ErrInvalidSubscription) + } + // PeriodSec is only required for PERIODIC mode + if newSubscription.Mode == ModePeriodic && newSubscription.PeriodSec <= 0 { + return "", fmt.Errorf("%w: PeriodSec must be > 0 for PERIODIC mode", ErrInvalidSubscription) + } + if newSubscription.NfID == "" { + return "", fmt.Errorf("%w: missing NfID", ErrInvalidSubscription) + } + if newSubscription.NotifyCorrelationID == "" { + return "", fmt.Errorf("%w: missing NotifyCorrelationID", ErrInvalidSubscription) + } + + subscriptionID := store.generateSubscriptionID() + now := time.Now() + + // initialize internal bookkeeping fields + newSubscription.ID = subscriptionID + newSubscription.CreatedAt = now + newSubscription.LastNotify = now // Initialize to now to avoid time overflow + if newSubscription.Snapshots == nil { + newSubscription.Snapshots = make(map[SessionKey]Counters) + } + + store.mutexForSubscriptions.Lock() + store.subscriptionsByID[subscriptionID] = newSubscription + store.mutexForSubscriptions.Unlock() + + return subscriptionID, nil +} + +// DeleteSubscription removes a subscription by ID. +func (store *SubscriptionStore) DeleteSubscription(subscriptionID string) error { + store.mutexForSubscriptions.Lock() + defer store.mutexForSubscriptions.Unlock() + + if _, ok := store.subscriptionsByID[subscriptionID]; !ok { + return ErrSubscriptionNotFound + } + delete(store.subscriptionsByID, subscriptionID) + return nil +} + +// GetSubscription returns the subscription pointer for the given ID and a boolean indicating presence. +// Callers should treat the returned pointer as read-mostly; mutations should be protected externally +// if done concurrently with other operations. +func (store *SubscriptionStore) GetSubscription(subscriptionID string) (*Subscription, bool) { + store.mutexForSubscriptions.RLock() + subscription, ok := store.subscriptionsByID[subscriptionID] + store.mutexForSubscriptions.RUnlock() + return subscription, ok +} + +// AllSubscriptions returns a snapshot slice of all current subscriptions. +// The returned slice is safe to iterate without holding the store lock. +func (store *SubscriptionStore) AllSubscriptions() []*Subscription { + store.mutexForSubscriptions.RLock() + defer store.mutexForSubscriptions.RUnlock() + + list := make([]*Subscription, 0, len(store.subscriptionsByID)) + for _, subscription := range store.subscriptionsByID { + list = append(list, subscription) + } + return list +} + +// generateSubscriptionID creates a sortable, mostly-unique ID composed of: +// - unix time in nanoseconds +// - a per-timestamp sequence (for same-ns collisions) +// - an optional instance suffix (configurable at store creation) +// +// Format example: "sub-1730023456789012345-0001-hostA" +func (store *SubscriptionStore) generateSubscriptionID() string { + nowNS := time.Now().UnixNano() + + store.mutexForIDGen.Lock() + if nowNS != store.lastTimestampNS { + store.lastTimestampNS = nowNS + store.sequencePerTS = 0 + } else { + store.sequencePerTS++ + } + sequenceCopy := store.sequencePerTS + instanceSuffixCopy := store.instanceSuffixStr + store.mutexForIDGen.Unlock() + + if instanceSuffixCopy != "" { + return fmt.Sprintf("sub-%d-%04d-%s", nowNS, sequenceCopy, instanceSuffixCopy) + } + return fmt.Sprintf("sub-%d-%04d", nowNS, sequenceCopy) +} diff --git a/internal/ees/types.go b/internal/ees/types.go new file mode 100644 index 00000000..cc1b967d --- /dev/null +++ b/internal/ees/types.go @@ -0,0 +1,221 @@ +// Package ees contains types and interfaces for the UPF Event Exposure Service (EES) +// MVP: USER_DATA_USAGE_MEASURES with per-PDU-session granularity and periodic/on-demand reporting. + +package ees + +import ( + "sync" + "time" +) + +// EventType enumerates EES event IDs supported by the UPF. +// MVP only: USER_DATA_USAGE_MEASURES. +type EventType string + +const ( + // EventUserDataUsageMeasures reports per-interval usage deltas (bytes/packets, UL/DL), + // optionally with throughputs derived from (delta bytes) / (EndTime - StartTime). + EventUserDataUsageMeasures EventType = "USER_DATA_USAGE_MEASURES" + // EventUserDataUsageTrends reports throughput statistics. + EventUserDataUsageTrends EventType = "USER_DATA_USAGE_TRENDS" +) + +// UserDataUsageMeasurements represents measurements per TS 29.564. +// Contains volume and/or throughput statistics measurements. +type UserDataUsageMeasurements struct { + // Granularity identifiers + AppId string `json:"appId,omitempty"` // For PER_APPLICATION granularity + FlowInfo *FlowInformation `json:"flowInfo,omitempty"` // For PER_FLOW granularity + + VolumeMeasurement *VolumeMeasurement `json:"volumeMeasurement,omitempty"` + ThroughputMeasurement *ThroughputMeasurement `json:"throughputMeasurement,omitempty"` + ThroughputStatisticsMeasurement *ThroughputStatisticsMeasurement `json:"throughputStatisticsMeasurement,omitempty"` +} + +// FlowInformation per TS 29.512 FlowInformation schema. +type FlowInformation struct { + FlowDescription string `json:"flowDescription,omitempty"` // IPFilterRule format + FlowDirection string `json:"flowDirection,omitempty"` // UPLINK, DOWNLINK, BIDIRECTIONAL +} + +// VolumeMeasurement per TS 29.564 schema. +type VolumeMeasurement struct { + TotalVolume uint64 `json:"totalVolume,omitempty"` + UlVolume uint64 `json:"ulVolume,omitempty"` + DlVolume uint64 `json:"dlVolume,omitempty"` + TotalNbOfPackets uint64 `json:"totalNbOfPackets,omitempty"` + UlNbOfPackets uint64 `json:"ulNbOfPackets,omitempty"` + DlNbOfPackets uint64 `json:"dlNbOfPackets,omitempty"` +} + +// ThroughputMeasurement per TS 29.564 schema. +type ThroughputMeasurement struct { + UlThroughput string `json:"ulThroughput,omitempty"` // BitRate + DlThroughput string `json:"dlThroughput,omitempty"` // BitRate + UlPacketThroughput string `json:"ulPacketThroughput,omitempty"` // PacketRate + DlPacketThroughput string `json:"dlPacketThroughput,omitempty"` // PacketRate +} + +// ThroughputStatisticsMeasurement per TS 29.564 schema. +type ThroughputStatisticsMeasurement struct { + UlAverageThroughput string `json:"ulAverageThroughput,omitempty"` // BitRate as string per spec + DlAverageThroughput string `json:"dlAverageThroughput,omitempty"` // BitRate as string per spec +} + +// Granularity controls the level of aggregation for the event. +// MVP only: perPduSession. +type Granularity string + +const ( + // GranularityPerSession reports measurements per PDU session. + GranularityPerSession Granularity = "PER_SESSION" + // GranularityPerApplication reports measurements grouped by application ID. + GranularityPerApplication Granularity = "PER_APPLICATION" + // GranularityPerFlow reports measurements grouped by traffic flow. + GranularityPerFlow Granularity = "PER_FLOW" +) + +// Mode controls how a subscription is served. +type Mode string + +const ( + // ModePeriodic: deliver reports every PeriodSec ticks. + ModePeriodic Mode = "PERIODIC" + // ModeOnDemand: deliver one immediate report using the current snapshot + // relative to the last snapshot (then refresh the snapshot and typically + // switch to PERIODIC in the aggregator per MVP behavior). + ModeOnDemand Mode = "ON_DEMAND" +) + +// MeasurementType specifies which measurements to include in reports. +// TS 29.564: Required when event type is USER_DATA_USAGE_MEASURES. +type MeasurementType string + +const ( + // MeasureVolume requests volume measurements (bytes/packets). + MeasureVolume MeasurementType = "VOLUME_MEASUREMENT" + // MeasureThroughput requests throughput measurements (bit rate). + MeasureThroughput MeasurementType = "THROUGHPUT_MEASUREMENT" + // MeasureAppInfo requests application-related information. + MeasureAppInfo MeasurementType = "APPLICATION_RELATED_INFO" +) + +// TargetScope defines the target selection for a subscription. +// MVP: only AnyUE=true (all active PDU sessions). +type TargetScope struct { + AnyUE bool + // Future: UEIP, SUPI, DNN, S-NSSAI, app filters... + // UeIPAddress allows targeting a specific UE by IP. + UeIPAddress string +} + +// SessionKey uniquely identifies a session for reporting purposes. +// MVP: use SEIDs; UE IP can be added later without breaking the map key. +type SessionKey struct { + LocalSEID uint64 // UPF user-plane SEID + RemoteSEID uint64 // SMF control-plane SEID + // Future: UEIP string +} + +// Counters represents raw counters collected for a single session over an interval. +// The interval is [StartTime, EndTime]; think "time window start/end" but field names +// are strictly StartTime / EndTime per naming guideline. +type Counters struct { + ULBytes uint64 + DLBytes uint64 + ULPackets uint64 + DLPackets uint64 + + StartTime time.Time // measurement interval start time + EndTime time.Time // measurement interval end time +} + +// UsageMeasures represents a delta (relative to last snapshot) plus derived metrics +// that are ready to be sent in a Notify payload. +type UsageMeasures struct { + Key SessionKey + + ULBytesDelta uint64 + DLBytesDelta uint64 + ULPacketsDelta uint64 + DLPacketsDelta uint64 + + StartTime time.Time // interval start + EndTime time.Time // interval end + + // Derived metrics (optional in MVP; aggregator may compute them). + ULThroughputBps float64 + DLThroughputBps float64 + + // Derived packet throughput (packets per second) + ULPacketThroughputPps float64 + DLPacketThroughputPps float64 + + // TS 29.564 UE identifiers + UeIpv4Addr string // UE IPv4 Address from session context +} + +// Subscription holds the in-memory state for a single EES subscription. +type Subscription struct { + mu sync.RWMutex + ID string + NotifURI string + NotifyCorrelationID string // Added: Client-provided correlation ID + NfID string // Added: NF Instance ID + Event EventType + Target TargetScope + Granularity Granularity + Mode Mode + PeriodSec int + + // TS 29.564: MeasurementTypes specifies which measurements are requested. + MeasurementTypes []MeasurementType + + // TS 29.564: Granularity-specific filters + AppIds []string `json:"appIds,omitempty"` // For PER_APPLICATION + TrafficFilters []FlowInformation `json:"trafficFilters,omitempty"` // For PER_FLOW + + CreatedAt time.Time + LastNotify time.Time + + // Snapshots keeps the last seen per-session counters for delta computation. + // Key: SessionKey (LocalSEID, RemoteSEID) + // Val: last counters over [StartTime, EndTime] + Snapshots map[SessionKey]Counters +} + +// HasMeasurementType checks if the subscription requests the given measurement type. +func (s *Subscription) HasMeasurementType(mt MeasurementType) bool { + for _, t := range s.MeasurementTypes { + if t == mt { + return true + } + } + return false +} + +// Source abstracts the producer of session-level counters for EES. +// SnapshotNow should return a *copy* of current per-session counters, each with +// their StartTime/EndTime representing the measurement interval the counters cover. +type Source interface { + SnapshotNow() (map[SessionKey]Counters, error) +} + +type PDRContext struct { + PDRID uint16 + URRIDs []uint32 +} + +type SessionContext struct { + RemoteSEID uint64 + UeIPv4Addr string // Added: UE IPv4 Address + URRIDs []uint32 + PDRs []*PDRContext +} + +// SessionProvider defines the interface for obtaining active Session information. +// Used by API Server for provisioning Shadow URRs to all active sessions. +type SessionProvider interface { + GetSessionContexts() map[uint64]SessionContext + GetSessionContextUEIP(lSeid uint64) (string, bool) +} diff --git a/internal/forwarder/gtp5g.go b/internal/forwarder/gtp5g.go index 4fc3a456..df880ad9 100644 --- a/internal/forwarder/gtp5g.go +++ b/internal/forwarder/gtp5g.go @@ -174,6 +174,12 @@ func (g *Gtp5g) Link() *Gtp5gLink { return g.link } +// GetPerioServer returns the periodic report server for querying URR periods. +// This is used by EES to validate subscription periods against URR measurement periods. +func (g *Gtp5g) GetPerioServer() *perio.Server { + return g.ps +} + func (g *Gtp5g) newFlowDesc(s string, swapSrcDst bool) (nl.AttrList, error) { var attrs nl.AttrList fd, err := ParseFlowDesc(s) @@ -256,16 +262,20 @@ func convertSlice(ports [][]uint16) []byte { func (g *Gtp5g) newSdfFilter(i *ie.IE, srcIf uint8) (nl.AttrList, error) { var attrs nl.AttrList + // i.Payload[0] corresponds to Octet 5 (Flags) in the spec flags := i.Payload[0] hasFD := (flags & 0x01) != 0 // Bit 1: Flow Description offset := 2 + if hasFD { if len(i.Payload) < offset+2 { return nil, errors.New("SDF Filter IE with FD flag needs Length of Flow Description") } // Read FDLength from i.Payload[2-3] (Octets 7-8 in spec) fdLength := uint16(i.Payload[offset])<<8 | uint16(i.Payload[offset+1]) + + // Validate FDLength doesn't exceed available payload // Flow Description data starts at i.Payload[4] (Octet 9) flowDescStart := offset + 2 availableBytes := len(i.Payload) - flowDescStart @@ -275,10 +285,13 @@ func (g *Gtp5g) newSdfFilter(i *ie.IE, srcIf uint8) (nl.AttrList, error) { fdLength, availableBytes) } } + + // Now it's safe to parse - the payload has been pre-validated v, err := i.SDFFilter() if err != nil { return nil, errors.Wrap(err, "failed to parse SDF Filter") } + // Process validated SDF Filter fields if v.HasFD() { swapSrcDst := (srcIf == ie.SrcInterfaceAccess) @@ -291,6 +304,7 @@ func (g *Gtp5g) newSdfFilter(i *ie.IE, srcIf uint8) (nl.AttrList, error) { Value: fd, }) } + if v.HasTTC() { // TODO: // v.ToSTrafficClass string @@ -382,6 +396,7 @@ func (g *Gtp5g) newPdi(i *ie.IE) (nl.AttrList, error) { // Validate SDF Filter IE payload length early (TS 29.244 Section 8.2.5) // Minimum: 1 byte (flags) + 1 byte (spare) + at least 1 byte for content if len(x.Payload) < 3 { + logger.FwderLog.Warnf("SDF Filter IE payload too short: %d bytes (minimum 3)", len(x.Payload)) return nil, errors.Errorf("SDF Filter IE payload too short: %d bytes (minimum 3)", len(x.Payload)) } sdfIEs = append(sdfIEs, x) @@ -392,6 +407,8 @@ func (g *Gtp5g) newPdi(i *ie.IE) (nl.AttrList, error) { for _, x := range sdfIEs { v, err := g.newSdfFilter(x, srcIf) if err != nil { + // Log the error and return it instead of silently ignoring + logger.FwderLog.Warnf("Failed to parse SDF Filter: %v", err) return nil, errors.Wrap(err, "newSdfFilter failed") } attrs = append(attrs, nl.Attr{ @@ -403,6 +420,200 @@ func (g *Gtp5g) newPdi(i *ie.IE) (nl.AttrList, error) { return attrs, nil } +func (g *Gtp5g) CreatePDR(lSeid uint64, req *ie.IE) error { + var pdrid uint64 + var attrs []nl.Attr + + ies, err := req.CreatePDR() + if err != nil { + return err + } + + for _, i := range ies { + switch i.Type { + case ie.PDRID: + v, err := i.PDRID() + if err != nil { + return errors.Wrap(err, "CreatePDR: failed to parse PDRID") + } + pdrid = uint64(v) + case ie.Precedence: + v, err := i.Precedence() + if err != nil { + return errors.Wrap(err, "CreatePDR: failed to parse Precedence") + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_PRECEDENCE, + Value: nl.AttrU32(v), + }) + case ie.PDI: + v, err := g.newPdi(i) + if err != nil { + return errors.Wrap(err, "CreatePDR: failed to parse PDI") + } + if v != nil { + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_PDI, + Value: v, + }) + } + case ie.OuterHeaderRemoval: + v, err := i.OuterHeaderRemovalDescription() + if err != nil { + return errors.Wrap(err, "CreatePDR: failed to parse OuterHeaderRemoval") + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_OUTER_HEADER_REMOVAL, + Value: nl.AttrU8(v), + }) + // ignore GTPUExternsionHeaderDeletion + case ie.FARID: + v, err := i.FARID() + if err != nil { + return errors.Wrap(err, "CreatePDR: failed to parse FARID") + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_FAR_ID, + Value: nl.AttrU32(v), + }) + case ie.QERID: + v, err := i.QERID() + if err != nil { + // QER is optional, log but continue + logger.FwderLog.Warnf("CreatePDR: Failed to parse QERID: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_QER_ID, + Value: nl.AttrU32(v), + }) + case ie.URRID: + v, err := i.URRID() + if err != nil { + // URR is optional, log but continue + logger.FwderLog.Warnf("CreatePDR: Failed to parse URRID: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_URR_ID, + Value: nl.AttrU32(v), + }) + } + } + + // TODO: + // Not in 3GPP spec, just used for routing + // var roleAddrIpv4 net.IP + // roleAddrIpv4 = net.IPv4(34, 35, 36, 37) + // pdr.RoleAddrIpv4 = &roleAddrIpv4 + + // TODO: + // Not in 3GPP spec, just used for buffering + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_UNIX_SOCKET_PATH, + Value: nl.AttrString(gtp5gnl.PdrAddrForNetlink), + }) + + oid := gtp5gnl.OID{lSeid, pdrid} + return gtp5gnl.CreatePDROID(g.client, g.link.link, oid, attrs) +} + +func (g *Gtp5g) UpdatePDR(lSeid uint64, req *ie.IE) error { + var pdrid uint64 + var attrs []nl.Attr + + ies, err := req.UpdatePDR() + if err != nil { + return err + } + + for _, i := range ies { + switch i.Type { + case ie.PDRID: + v, err := i.PDRID() + if err != nil { + return errors.Wrap(err, "UpdatePDR: failed to parse PDRID") + } + pdrid = uint64(v) + case ie.Precedence: + v, err := i.Precedence() + if err != nil { + // Precedence is optional in Update, log but continue + logger.FwderLog.Warnf("UpdatePDR: Failed to parse Precedence: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_PRECEDENCE, + Value: nl.AttrU32(v), + }) + case ie.PDI: + v, err := g.newPdi(i) + if err != nil { + return errors.Wrap(err, "UpdatePDR: failed to parse PDI") + } + if v != nil { + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_PDI, + Value: v, + }) + } + case ie.OuterHeaderRemoval: + v, err := i.OuterHeaderRemovalDescription() + if err != nil { + logger.FwderLog.Warnf("UpdatePDR: Failed to parse OuterHeaderRemoval: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_OUTER_HEADER_REMOVAL, + Value: nl.AttrU8(v), + }) + // ignore GTPUExternsionHeaderDeletion + case ie.FARID: + v, err := i.FARID() + if err != nil { + logger.FwderLog.Warnf("UpdatePDR: Failed to parse FARID: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_FAR_ID, + Value: nl.AttrU32(v), + }) + case ie.QERID: + v, err := i.QERID() + if err != nil { + logger.FwderLog.Warnf("UpdatePDR: Failed to parse QERID: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_QER_ID, + Value: nl.AttrU32(v), + }) + case ie.URRID: + v, err := i.URRID() + if err != nil { + logger.FwderLog.Warnf("UpdatePDR: Failed to parse URRID: %v", err) + break + } + attrs = append(attrs, nl.Attr{ + Type: gtp5gnl.PDR_URR_ID, + Value: nl.AttrU32(v), + }) + } + } + + oid := gtp5gnl.OID{lSeid, pdrid} + return gtp5gnl.UpdatePDROID(g.client, g.link.link, oid, attrs) +} + +func (g *Gtp5g) RemovePDR(lSeid uint64, req *ie.IE) error { + v, err := req.PDRID() + if err != nil { + return errors.New("not found PDRID") + } + oid := gtp5gnl.OID{lSeid, uint64(v)} + return gtp5gnl.RemovePDROID(g.client, g.link.link, oid) +} + func (g *Gtp5g) newForwardingParameter(ies []*ie.IE) (nl.AttrList, error) { var attrs nl.AttrList @@ -1546,8 +1757,8 @@ func (g *Gtp5g) BuildUpdateURRPlan(lSeid uint64, req *ie.IE) (*URRPlan, error) { Value: v, }) case ie.VolumeQuota: - v, err := g.newVolumeQuota(i) - if err != nil { + v, err1 := g.newVolumeQuota(i) + if err1 != nil { break } attrs = append(attrs, nl.Attr{ @@ -1588,7 +1799,7 @@ func (g *Gtp5g) BuildRemoveURRPlan(lSeid uint64, req *ie.IE) (*URRPlan, error) { func (g *Gtp5g) BuildQueryURRPlan(lSeid uint64, req *ie.IE) (*URRPlan, error) { v, err := req.URRID() if err != nil { - return nil, errors.New("not found URRID") + return nil, err } return &URRPlan{ diff --git a/internal/forwarder/perio/server.go b/internal/forwarder/perio/server.go index 1bcc38f7..ba9d9340 100644 --- a/internal/forwarder/perio/server.go +++ b/internal/forwarder/perio/server.go @@ -89,11 +89,15 @@ func (pg *PERIOGroup) stopTicker() { } type Server struct { + mu sync.RWMutex evtCh chan Event perioList map[time.Duration]*PERIOGroup // key: period handler report.Handler queryURR func(map[uint64][]uint32) (map[uint64][]report.USAReport, error) + + // Callback when a new URR is added (for EES period adjustment) + onURRAdded func(urrid uint32, period time.Duration) } func OpenServer(wg *sync.WaitGroup) (*Server, error) { @@ -120,6 +124,12 @@ func (s *Server) Handle( s.queryURR = queryURR } +// SetOnURRAdded sets the callback function that is invoked when a new URR is added. +// This is used by EES to trigger automatic period adjustment. +func (s *Server) SetOnURRAdded(callback func(urrid uint32, period time.Duration)) { + s.onURRAdded = callback +} + func (s *Server) Serve(wg *sync.WaitGroup) { logger.PerioLog.Infof("perio server started") defer func() { @@ -132,6 +142,7 @@ func (s *Server) Serve(wg *sync.WaitGroup) { logger.PerioLog.Infof("recv event[%s][%+v]", e.eType, e) switch e.eType { case TYPE_PERIO_ADD: + s.mu.Lock() perioGroup, ok := s.perioList[e.period] if !ok { // New ticker if no this period ticker found @@ -142,11 +153,14 @@ func (s *Server) Serve(wg *sync.WaitGroup) { err := perioGroup.newTicker(wg, s.evtCh) if err != nil { logger.PerioLog.Errorln(err) + s.mu.Unlock() continue } s.perioList[e.period] = perioGroup } + cb := s.onURRAdded + urrids := perioGroup.urrids[e.lSeid] if urrids == nil { perioGroup.urrids[e.lSeid] = make(map[uint32]struct{}) @@ -157,7 +171,14 @@ func (s *Server) Serve(wg *sync.WaitGroup) { perioGroup.urrids[e.lSeid][e.urrid] = struct{}{} } } + s.mu.Unlock() + + // Trigger callback for EES period adjustment outside the lock + if cb != nil { + cb(e.urrid, e.period) + } case TYPE_PERIO_DEL: + s.mu.Lock() for period, perioGroup := range s.perioList { _, ok := perioGroup.urrids[e.lSeid][e.urrid] if ok { @@ -174,11 +195,14 @@ func (s *Server) Serve(wg *sync.WaitGroup) { break } } + s.mu.Unlock() case TYPE_PERIO_TIMEOUT: var lSeidUrridsMap map[uint64][]uint32 + s.mu.RLock() perioGroup, ok := s.perioList[e.period] if !ok { + s.mu.RUnlock() logger.PerioLog.Warnf("no periodGroup found for period[%v]", e.period) break } @@ -189,6 +213,7 @@ func (s *Server) Serve(wg *sync.WaitGroup) { lSeidUrridsMap[lSeid] = append(lSeidUrridsMap[lSeid], urrId) } } + s.mu.RUnlock() seidUsars, err := s.queryURR(lSeidUrridsMap) if err != nil { @@ -215,10 +240,12 @@ func (s *Server) Serve(wg *sync.WaitGroup) { }) } case TYPE_SERVER_CLOSE: + s.mu.Lock() for period, perioGroup := range s.perioList { perioGroup.stopTicker() delete(s.perioList, period) } + s.mu.Unlock() return } } @@ -240,3 +267,20 @@ func (s *Server) DelPeriodReportTimer(lSeid uint64, urrid uint32) { urrid: urrid, } } + +// GetAnyURRPeriod returns the period of any active URR with the specified ID +// across all sessions. Useful for getting a representative period when the +// exact session is unknown. Returns 0 if the URR is not found. +func (s *Server) GetAnyURRPeriod(urrid uint32) time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + for period, perioGroup := range s.perioList { + for _, urrids := range perioGroup.urrids { + _, ok := urrids[urrid] + if ok { + return period // Found an instance of this URR + } + } + } + return 0 +} diff --git a/internal/logger/logger.go b/internal/logger/logger.go index dc580d1c..28cbff1a 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -15,6 +15,7 @@ var ( BuffLog *logrus.Entry PerioLog *logrus.Entry FwderLog *logrus.Entry + EesLog *logrus.Entry ) func init() { @@ -36,4 +37,5 @@ func init() { BuffLog = NfLog.WithField(logger_util.FieldCategory, "BUFF") PerioLog = NfLog.WithField(logger_util.FieldCategory, "Perio") FwderLog = NfLog.WithField(logger_util.FieldCategory, "FWD") + EesLog = NfLog.WithField(logger_util.FieldCategory, "EES") } diff --git a/internal/pfcp/node.go b/internal/pfcp/node.go index ce793fdf..c997ed41 100644 --- a/internal/pfcp/node.go +++ b/internal/pfcp/node.go @@ -3,11 +3,13 @@ package pfcp import ( "fmt" "net" + "sync" "github.com/pkg/errors" "github.com/sirupsen/logrus" "github.com/wmnsk/go-pfcp/ie" + "github.com/free5gc/go-upf/internal/ees" "github.com/free5gc/go-upf/internal/forwarder" "github.com/free5gc/go-upf/internal/report" logger_util "github.com/free5gc/util/logger" @@ -30,17 +32,18 @@ type URRInfo struct { } type Sess struct { - rnode *RemoteNode - LocalID uint64 - RemoteID uint64 - PDRIDs map[uint16]*PDRInfo // key: PDR_ID - FARIDs map[uint32]struct{} // key: FAR_ID - QERIDs map[uint32]struct{} // key: QER_ID - URRIDs map[uint32]*URRInfo // key: URR_ID - BARIDs map[uint8]struct{} // key: BAR_ID - q map[uint16]chan []byte // key: PDR_ID - qlen int - log *logrus.Entry + rnode *RemoteNode + LocalID uint64 + RemoteID uint64 + UeIPv4Addr string // Added: UE IPv4 Address derived from PDR + PDRIDs map[uint16]*PDRInfo // key: PDR_ID + FARIDs map[uint32]struct{} // key: FAR_ID + QERIDs map[uint32]struct{} // key: QER_ID + URRIDs map[uint32]*URRInfo // key: URR_ID + BARIDs map[uint8]struct{} // key: BAR_ID + q map[uint16]chan []byte // key: PDR_ID + qlen int + log *logrus.Entry } var ( @@ -107,15 +110,6 @@ func (s *Sess) Close() []report.USAReport { s.log.Errorf("Execute Deletion Plan err: %v", err) } - // Apply state changes and collect USAReports - var usars []report.USAReport - - for _, p := range plan.RemovePDRs { - rs := s.ApplyRemovePDR(p) - if len(rs) > 0 { - usars = append(usars, rs...) - } - } for _, p := range plan.RemoveBARs { s.ApplyRemoveBAR(p) } @@ -129,6 +123,7 @@ func (s *Sess) Close() []report.USAReport { s.ApplyRemoveFAR(p) } + var usars []report.USAReport // Collect USAReports from execution result (RemoveURR) if execResult != nil && len(execResult.USAReports) > 0 { for i := range execResult.USAReports { @@ -673,9 +668,13 @@ func (n *RemoteNode) DeleteSess(lSeid uint64) []report.USAReport { type LocalNode struct { sess []*Sess free []uint64 + mu sync.RWMutex } func (n *LocalNode) Reset() { + n.mu.Lock() + defer n.mu.Unlock() + for _, sess := range n.sess { if sess != nil { sess.Close() @@ -690,6 +689,9 @@ func (n *LocalNode) Sess(lSeid uint64) (*Sess, error) { return nil, errors.New("Sess: invalid lSeid:0") } + n.mu.RLock() + defer n.mu.RUnlock() + // Length as int; compare as uint64 to match lSeid type. sessLen := len(n.sess) if lSeid > uint64(sessLen) { @@ -706,6 +708,9 @@ func (n *LocalNode) Sess(lSeid uint64) (*Sess, error) { } func (n *LocalNode) RemoteSess(rSeid uint64, addr net.Addr) (*Sess, error) { + n.mu.RLock() + defer n.mu.RUnlock() + addrString := "" if addr != nil { addrString = addr.String() @@ -723,6 +728,9 @@ func (n *LocalNode) RemoteSess(rSeid uint64, addr net.Addr) (*Sess, error) { } func (n *LocalNode) NewSess(rSeid uint64, qlen int) *Sess { + n.mu.Lock() + defer n.mu.Unlock() + s := &Sess{ RemoteID: rSeid, PDRIDs: make(map[uint16]*PDRInfo), @@ -750,6 +758,9 @@ func (n *LocalNode) DeleteSess(lSeid uint64) ([]report.USAReport, error) { return nil, errors.New("DeleteSess: invalid lSeid:0") } + n.mu.Lock() + defer n.mu.Unlock() + // Capacity as int; compare as uint64 to match lSeid type. sessCap := len(n.sess) if lSeid > uint64(sessCap) { @@ -769,3 +780,65 @@ func (n *LocalNode) DeleteSess(lSeid uint64) ([]report.USAReport, error) { return usars, nil } + +// GetSessionContextUEIP returns the UE IPv4 address for a given SEID. +// This is more efficient than GetSessionContexts when only the IP is needed. +func (n *LocalNode) GetSessionContextUEIP(lSeid uint64) (string, bool) { + n.mu.RLock() + defer n.mu.RUnlock() + + idx := int(lSeid) - 1 + if idx < 0 || idx >= len(n.sess) || n.sess[idx] == nil { + return "", false + } + return n.sess[idx].UeIPv4Addr, true +} + +// [New] Implement ees.SessionProvider interface +// Allow EES to obtain the context of active Sessions (RemoteSEID and URRIDs) +func (n *LocalNode) GetSessionContexts() map[uint64]ees.SessionContext { + n.mu.RLock() + defer n.mu.RUnlock() + + result := make(map[uint64]ees.SessionContext) + + for _, sess := range n.sess { + if sess == nil { + continue + } + + // Collect all URR IDs under this Session + var urrIDs []uint32 + if sess.URRIDs != nil { + for urrID := range sess.URRIDs { + urrIDs = append(urrIDs, urrID) + } + } + + // Only monitor when Session has URR (or decide whether to include empty Session based on requirements) + if len(urrIDs) > 0 { + // Populate PDRs + var pdrs []*ees.PDRContext + + // Correct iteration over PDRIDs map + for pdrID, pdrInfo := range sess.PDRIDs { + var pdrURRIDs []uint32 + for uid := range pdrInfo.RelatedURRIDs { + pdrURRIDs = append(pdrURRIDs, uid) + } + pdrs = append(pdrs, &ees.PDRContext{ + PDRID: pdrID, + URRIDs: pdrURRIDs, + }) + } + + result[sess.LocalID] = ees.SessionContext{ + RemoteSEID: sess.RemoteID, + UeIPv4Addr: sess.UeIPv4Addr, + URRIDs: urrIDs, + PDRs: pdrs, + } + } + } + return result +} diff --git a/internal/pfcp/pfcp.go b/internal/pfcp/pfcp.go index 05daab85..e3227bfc 100644 --- a/internal/pfcp/pfcp.go +++ b/internal/pfcp/pfcp.go @@ -408,6 +408,10 @@ func setReqSeq(msgtmp message.Message, seq uint32) { } } +func (s *PfcpServer) GetLocalNode() *LocalNode { + return &s.lnode +} + func validatePfcpPacketLength(buf []byte) error { // Minimum PFCP message: 8 bytes (header without SEID) if len(buf) < 8 { diff --git a/internal/pfcp/report.go b/internal/pfcp/report.go index 31dd3066..f61699f8 100644 --- a/internal/pfcp/report.go +++ b/internal/pfcp/report.go @@ -105,6 +105,7 @@ func (s *PfcpServer) serveUSAReport(addr net.Addr, lSeid uint64, usars []report. 0, ie.NewReportType(0, 0, 1, 0), ) + s.log.Infof("number of usars is : %d", len(usars)) for _, r := range usars { urrInfo, ok := sess.URRIDs[r.URRID] if !ok { @@ -112,6 +113,7 @@ func (s *PfcpServer) serveUSAReport(addr net.Addr, lSeid uint64, usars []report. continue } r.URSEQN = sess.URRSeq(r.URRID) + req.UsageReport = append(req.UsageReport, ie.NewUsageReportWithinSessionReportRequest( r.IEsWithinSessReportReq( diff --git a/internal/pfcp/session.go b/internal/pfcp/session.go index 25d0d3bb..6a1f9dd0 100644 --- a/internal/pfcp/session.go +++ b/internal/pfcp/session.go @@ -530,6 +530,18 @@ func (s *PfcpServer) handleSessionModificationRequest( )) } + // [New] Push collected USAReports to the internal Dispatcher so EES can receive them! + if len(usars) > 0 { + var rpts []report.Report + for i := range usars { + rpts = append(rpts, usars[i]) + } + s.NotifySessReport(report.SessReport{ + SEID: sess.RemoteID, + Reports: rpts, + }) + } + // Cleanup removed URRs sess.CleanupRemovedURRs() @@ -598,6 +610,18 @@ func (s *PfcpServer) handleSessionDeletionRequest( } } + // [New] Push collected USAReports to the internal Dispatcher so EES can receive them! + if len(usars) > 0 { + var rpts []report.Report + for i := range usars { + rpts = append(rpts, usars[i]) + } + s.NotifySessReport(report.SessReport{ + SEID: sess.RemoteID, + Reports: rpts, + }) + } + err = s.sendRspTo(rsp, addr) if err != nil { s.log.Errorln(err) diff --git a/pkg/app/app.go b/pkg/app/app.go index f0bce85d..8dc05368 100644 --- a/pkg/app/app.go +++ b/pkg/app/app.go @@ -2,15 +2,19 @@ package app import ( "context" + "fmt" "os" "os/signal" "runtime/debug" "sync" "syscall" + "time" "github.com/sirupsen/logrus" + "github.com/free5gc/go-upf/internal/ees" "github.com/free5gc/go-upf/internal/forwarder" + "github.com/free5gc/go-upf/internal/forwarder/perio" "github.com/free5gc/go-upf/internal/logger" "github.com/free5gc/go-upf/internal/pfcp" "github.com/free5gc/go-upf/pkg/factory" @@ -22,6 +26,7 @@ type UpfApp struct { cfg *factory.Config driver forwarder.Driver pfcpServer *pfcp.PfcpServer + eesServer *ees.Server } func NewApp(cfg *factory.Config) (*UpfApp, error) { @@ -61,15 +66,97 @@ func (a *UpfApp) SetLogReportCaller(reportCaller bool) { logger.Log.SetReportCaller(reportCaller) } +/* + func (u *UpfApp) Run() error { + var cancel context.CancelFunc + u.ctx, cancel = context.WithCancel(context.Background()) + defer cancel() + + u.wg.Add(1) + // Go Routine is spawned here for listening for cancellation event on + // context + go u.listenShutdownEvent() + + var err error + u.driver, err = forwarder.NewDriver(&u.wg, u.cfg) + if err != nil { + return err + } + + u.pfcpServer = pfcp.NewPfcpServer(u.cfg, u.driver) + u.driver.HandleReport(u.pfcpServer) + u.pfcpServer.Start(&u.wg) + + logger.MainLog.Infoln("UPF started") + + // Wait for interrupt signal to gracefully shutdown + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + <-sigCh + + // Receive the interrupt signal + logger.MainLog.Infof("Shutdown UPF ...") + // Notify each goroutine and wait them stopped + cancel() + u.WaitRoutineStopped() + logger.MainLog.Infof("UPF exited") + return nil + } +*/ +func (u *UpfApp) listenShutdownEvent() { + defer func() { + if p := recover(); p != nil { + // Print stack for panic to log. Fatalf() will let program exit. + logger.MainLog.Fatalf("panic: %v\n%s", p, string(debug.Stack())) + } + + u.wg.Done() + }() + + <-u.ctx.Done() + if u.pfcpServer != nil { + u.pfcpServer.Stop() + } + if u.eesServer != nil { + // Use a short timeout for API server shutdown + shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := u.eesServer.Shutdown(shutdownCtx); err != nil { + logger.MainLog.Errorf("EES API Server Shutdown Error: %v", err) + } + } + if u.driver != nil { + u.driver.Close() + } +} + +func (u *UpfApp) WaitRoutineStopped() { + u.wg.Wait() + u.Terminate() +} + +func (u *UpfApp) Start() { + if err := u.Run(); err != nil { + logger.MainLog.Errorf("UPF Run err: %v", err) + } +} + +func (u *UpfApp) Terminate() { + logger.MainLog.Infof("Terminating UPF...") + logger.MainLog.Infof("UPF terminated") +} + func (u *UpfApp) Run() error { var cancel context.CancelFunc + u.ctx, cancel = context.WithCancel(context.Background()) defer cancel() u.wg.Add(1) - /* Go Routine is spawned here for listening for cancellation event on - * context */ + // Go Routine is spawned here for listening for cancellation event on + // context go u.listenShutdownEvent() + // ... (Original context setting) ... var err error u.driver, err = forwarder.NewDriver(&u.wg, u.cfg) @@ -78,11 +165,21 @@ func (u *UpfApp) Run() error { } u.pfcpServer = pfcp.NewPfcpServer(u.cfg, u.driver) - u.driver.HandleReport(u.pfcpServer) + + // [New] Dispatcher wiring + reportDispatcher := NewDispatcher(u.pfcpServer) + u.driver.HandleReport(reportDispatcher) + u.pfcpServer.Start(&u.wg) + // [New] Initialize EES if enabled + if err := u.initEES(reportDispatcher); err != nil { + return err + } + logger.MainLog.Infoln("UPF started") + // ... (Subsequent Signal handling remains unchanged) // Wait for interrupt signal to gracefully shutdown sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) @@ -97,37 +194,79 @@ func (u *UpfApp) Run() error { return nil } -func (u *UpfApp) listenShutdownEvent() { - defer func() { - if p := recover(); p != nil { - // Print stack for panic to log. Fatalf() will let program exit. - logger.MainLog.Fatalf("panic: %v\n%s", p, string(debug.Stack())) - } +func (u *UpfApp) initEES(reportDispatcher *Dispatcher) error { + if u.cfg.EES == nil || !u.cfg.EES.Enabled { + return nil + } - u.wg.Done() - }() + logger.MainLog.Infoln("Starting EES Module (Pure Push Mode)...") - <-u.ctx.Done() - if u.pfcpServer != nil { - u.pfcpServer.Stop() + // 1. Use existing EesLog + eesLogger := logger.EesLog + + // 2. Create Store / Notifier / Aggregator + subscriptionStore := ees.NewSubscriptionStore("") + notifier := ees.NewNotifier(eesLogger) + + period := 10 + if u.cfg.EES.PeriodSec > 0 { + period = u.cfg.EES.PeriodSec } - if u.driver != nil { - u.driver.Close() + + // Pure Push mode: Reports come from kernel + localNode := u.pfcpServer.GetLocalNode() + sessionProvider := localNode + + // Get perioServer from driver for URR period queries + var perioServer *perio.Server + if gtp5gDriver, ok := u.driver.(*forwarder.Gtp5g); ok { + perioServer = gtp5gDriver.GetPerioServer() } -} -func (u *UpfApp) WaitRoutineStopped() { - u.wg.Wait() - u.Terminate() -} + aggregator := ees.NewAggregator( + subscriptionStore, + time.Duration(period)*time.Second, + notifier, + eesLogger, + sessionProvider, + perioServer, // Pass perioServer for period validation + ) -func (u *UpfApp) Start() { - if err := u.Run(); err != nil { - logger.MainLog.Errorf("UPF Run err: %v", err) + // 3. Register EES Handler to Dispatcher + eesHandler := ees.NewHandler(aggregator, eesLogger) + reportDispatcher.RegisterEESHandler(eesHandler) + + // Set perioServer for dispatcher callbacks + if perioServer != nil { + reportDispatcher.SetPerioServer(perioServer) + + // Register callback to adjust aggregator period when URR is added + perioServer.SetOnURRAdded(func(urrid uint32, period time.Duration) { + // Only adjust for URR 2 (the periodic measurement URR) + if urrid == 2 { + logger.MainLog.Infof("EES: URR %d added with period %v, triggering aggregator adjustment", urrid, period) + aggregator.AdjustReportPeriod(period) + } + }) } -} -func (u *UpfApp) Terminate() { - logger.MainLog.Infof("Terminating UPF...") - logger.MainLog.Infof("UPF terminated") + // 4. Start Aggregator (processes buffered reports periodically) + go aggregator.Run(u.ctx) + + // 5. Start API Server (simplified - no Shadow URR provisioning) + port := u.cfg.EES.Port + if port == 0 { + port = 8088 + } + listenAddr := fmt.Sprintf("%s:%d", u.cfg.EES.IP, port) + u.eesServer = ees.NewServer(subscriptionStore, aggregator, eesLogger) + + go func() { + if err := u.eesServer.Serve(listenAddr); err != nil { + logger.MainLog.Errorf("EES API Server Error: %v", err) + } + }() + + logger.MainLog.Infof("EES started at %s with period %ds (Pure Push Mode - SMF URR)", listenAddr, period) + return nil } diff --git a/pkg/app/dispatcher.go b/pkg/app/dispatcher.go new file mode 100644 index 00000000..b4a419c6 --- /dev/null +++ b/pkg/app/dispatcher.go @@ -0,0 +1,65 @@ +package app + +import ( + "time" + + "github.com/free5gc/go-upf/internal/report" +) + +// Dispatcher implements report.Handler and multicasts reports to registered handlers. +// It allows both PFCP (SMF) and EES to receive usage reports. +type Dispatcher struct { + // Primary: PFCP Server (also handles buffering) + pfcpHandler report.Handler + // Secondary: EES Handler (optional, only receives reports) + eesHandler report.Handler + + // PerioServer interface (for querying URR periods) + perioServer interface { + GetAnyURRPeriod(urrid uint32) time.Duration + } +} + +// NewDispatcher creates a dispatcher. +func NewDispatcher(pfcpHandler report.Handler) *Dispatcher { + return &Dispatcher{ + pfcpHandler: pfcpHandler, + } +} + +// RegisterEESHandler registers the EES handler. +func (d *Dispatcher) RegisterEESHandler(handler report.Handler) { + d.eesHandler = handler +} + +// NotifySessReport multicasts the report to all registered handlers. +// Pure Push Mode: All USAReports are forwarded to both handlers. +// - PFCP handler: forwards to SMF (N4) +// - EES handler: aggregates for event exposure (filters URRID == 2 internally) +func (d *Dispatcher) NotifySessReport(sessRpt report.SessReport) { + // Dispatch to PFCP (N4) - all reports + if d.pfcpHandler != nil { + d.pfcpHandler.NotifySessReport(sessRpt) + } + + // Dispatch to EES - all reports (EES aggregator filters by URRID == 2) + if d.eesHandler != nil { + d.eesHandler.NotifySessReport(sessRpt) + } +} + +// PopBufPkt delegates buffering logic exclusively to the PFCP handler. +// EES does not handle buffering. +func (d *Dispatcher) PopBufPkt(seid uint64, pdrid uint16) ([]byte, bool) { + if d.pfcpHandler != nil { + return d.pfcpHandler.PopBufPkt(seid, pdrid) + } + return nil, false +} + +// SetPerioServer stores the perio server reference for URR period queries. +func (d *Dispatcher) SetPerioServer(perioServer interface{}) { + if ps, ok := perioServer.(interface{ GetAnyURRPeriod(uint32) time.Duration }); ok { + d.perioServer = ps + } +} diff --git a/pkg/factory/config.go b/pkg/factory/config.go index 81bf3b22..b5ccd62c 100644 --- a/pkg/factory/config.go +++ b/pkg/factory/config.go @@ -16,12 +16,13 @@ const ( ) type Config struct { - Version string `yaml:"version" valid:"required,in(1.0.3)"` - Description string `yaml:"description" valid:"optional"` - Pfcp *Pfcp `yaml:"pfcp" valid:"required"` - Gtpu *Gtpu `yaml:"gtpu" valid:"required"` - DnnList []DnnList `yaml:"dnnList" valid:"required"` - Logger *Logger `yaml:"logger" valid:"required"` + Version string `yaml:"version" valid:"required,in(1.0.3)"` + Description string `yaml:"description" valid:"optional"` + Pfcp *Pfcp `yaml:"pfcp" valid:"required"` + Gtpu *Gtpu `yaml:"gtpu" valid:"required"` + DnnList []DnnList `yaml:"dnnList" valid:"required"` + Logger *Logger `yaml:"logger" valid:"required"` + EES *EESConfig `yaml:"ees" valid:"optional"` } type Pfcp struct { @@ -56,6 +57,13 @@ type Logger struct { ReportCaller bool `yaml:"reportCaller" valid:"optional"` } +type EESConfig struct { + Enabled bool `yaml:"enabled" valid:"optional"` + IP string `yaml:"ip" valid:"optional"` + Port int `yaml:"port" valid:"optional"` + PeriodSec int `yaml:"periodSec" valid:"optional"` +} + func (c *Config) GetVersion() string { return c.Version }