diff --git a/drsm/claim.go b/drsm/claim.go index 523a847..0d99812 100644 --- a/drsm/claim.go +++ b/drsm/claim.go @@ -12,16 +12,38 @@ import ( func (d *Drsm) podDownDetected() { logger.DrsmLog.Infoln("started Pod Down goroutine") + for p := range d.podDown { logger.DrsmLog.Infof("pod Down detected %v", p) - // Given Pod find out current Chunks owned by this POD - pd := d.podMap[p] + + // Safely get pod entry + d.podMapMutex.RLock() + pd, found := d.podMap[p] + d.podMapMutex.RUnlock() + + if !found || pd == nil { + logger.DrsmLog.Warnf("pod %s not found in podMap", p) + continue + } + + // Copy chunk IDs while holding read lock + pd.mu.RLock() + chunkIDs := make([]int32, 0, len(pd.podChunks)) for k := range pd.podChunks { - d.globalChunkTblMutex.Lock() + chunkIDs = append(chunkIDs, k) + } + pd.mu.RUnlock() + + // Process chunks without holding pod lock + for _, k := range chunkIDs { + + d.globalChunkTblMutex.RLock() c, found := d.globalChunkTbl[k] - d.globalChunkTblMutex.Unlock() + d.globalChunkTblMutex.RUnlock() + logger.DrsmLog.Debugf("found: %v chunk: %v", found, c) - if found { + + if found && c != nil { go c.claimChunk(d, pd.PodId.PodName) } } diff --git a/drsm/drsm.go b/drsm/drsm.go index 8068e9a..d296d28 100644 --- a/drsm/drsm.go +++ b/drsm/drsm.go @@ -35,10 +35,12 @@ type chunk struct { } type podData struct { - PodId PodId `bson:"podId,omitempty" json:"podId,omitempty"` - Timestamp time.Time `bson:"time,omitempty" json:"time,omitempty"` - PrevTimestamp time.Time `bson:"-" json:"-"` - podChunks map[int32]*chunk `bson:"-" json:"-"` // chunkId to Chunk + PodId PodId + Timestamp time.Time + PrevTimestamp time.Time + + mu sync.RWMutex + podChunks map[int32]*chunk } type Drsm struct { @@ -77,14 +79,21 @@ func (d *Drsm) ConstuctDrsm(opt *Options) { } d.resourceValidCb = opt.ResourceValidCb } + d.chunkIdRange = 1 << (d.resIdSize - 10) logger.DrsmLog.Debugf("chunkId in the range of 0 to %v", d.chunkIdRange) + d.localChunkTbl = make(map[int32]*chunk) d.globalChunkTbl = make(map[int32]*chunk) d.podMap = make(map[string]*podData) - d.podDown = make(chan string, 10) d.scanChunks = make(map[int32]*chunk) - d.globalChunkTblMutex = sync.Mutex{} + + d.podDown = make(chan string, 10) + + // No explicit initialization required, but fine to keep + d.globalChunkTblMutex = sync.RWMutex{} + d.podMapMutex = sync.RWMutex{} + d.initIpam(opt) // connect to DB diff --git a/httpwrapper/httpwrapper.go b/httpwrapper/httpwrapper.go index 7a40404..050acd3 100644 --- a/httpwrapper/httpwrapper.go +++ b/httpwrapper/httpwrapper.go @@ -12,6 +12,7 @@ import ( "os" "time" + "github.com/5GC-DEV/util-cdac/logger" "github.com/pkg/errors" "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" @@ -57,10 +58,17 @@ func NewHttp2Server(bindAddr string, preMasterSecretLogPath string, handler http return nil, errors.New("server needs handler to handle request") } - h2Server := &http2.Server{ + /*h2Server := &http2.Server{ // TODO: extends the idle time after re-use openapi client IdleTimeout: 1 * time.Millisecond, + }*/ + h2Server := &http2.Server{ + IdleTimeout: 60 * time.Second, } + logger.UtilLog.Infof( + "HTTP2 IdleTimeout=%v", + h2Server.IdleTimeout, + ) server := &http.Server{ Addr: bindAddr, Handler: h2c.NewHandler(handler, h2Server), diff --git a/middleware/3gpp_Sbi_Request_Info.go b/middleware/3gpp_Sbi_Request_Info.go new file mode 100644 index 0000000..90061b1 --- /dev/null +++ b/middleware/3gpp_Sbi_Request_Info.go @@ -0,0 +1,49 @@ +package middleware + +import ( + "net/http" + "sync" + + "github.com/5GC-DEV/util-cdac/logger" + "github.com/gin-gonic/gin" +) + +var requestCache sync.Map + +func IdempotencyMiddleware() gin.HandlerFunc { + return func(c *gin.Context) { + + if c.Request.Method != http.MethodPost { + c.Next() + return + } + + requestInfo := c.GetHeader("3gpp-Sbi-Request-Info") + + if requestInfo != "" { + logger.UtilLog.Infof( + "Received idempotency key=%s method=%s path=%s", + requestInfo, + c.Request.Method, + c.Request.URL.Path, + ) + } + + if requestInfo == "" { + c.Next() + return + } + + if _, exists := requestCache.Load(requestInfo); exists { + c.JSON(http.StatusConflict, gin.H{ + "error": "duplicate request detected", + }) + c.Abort() + return + } + + requestCache.Store(requestInfo, true) + + c.Next() + } +}