Loading go-apps/meep-app-enablement/server/edge-3/eas_lifecycle.go 0 → 100644 +457 −0 Changes for go-apps/meep-app-enablement/server/edge-3/eas_lifecycle.go: 457 added lines, 0 removed lines. Original line number Diff line number Diff line /* * Copyright (c) 2026 The AdvantEDGE Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // EAS registration lifecycle (3GPP TS 29.558 Eees_EASRegistration): create, read, update, modify, // delete and expiry of the EAS registrations of this MEP. // // Every registration is mirrored into the MEC platform through the adapter (app instance, MEC 011 // registration and service), so that native MEC applications see the EAS over Mp1. The registration // record (EAS registration as granted + the MEC platform resource ids) is kept in the EDGE-3 store, // which also makes the expiry timers restart-safe. package server import ( "context" "encoding/json" "errors" "io/ioutil" "net/http" "strings" "sync" "time" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" "github.com/google/uuid" "github.com/gorilla/mux" ) // ProblemDetails cause of a second registration of the same EAS on this MEP (sandbox-defined) const causeEasAlreadyRegistered = "EAS_ALREADY_REGISTERED" // Reconciliation of stored registrations after a restart: retried until the MEC platform answers var reconcileRetryInterval = 5 * time.Second var reconcileTimeout = 5 * time.Minute // easRecord - EAS registration record of the EDGE-3 store type easRecord struct { Registration EasRegistration `json:"registration"` Links easLinks `json:"links"` } // easMutex serializes the EAS registration operations of this MEP. It is never held while waiting for // the app-enablement global mutex (the MEC 011 handlers take that one). var easMutex sync.Mutex var expiryTimers = map[string]*time.Timer{} var basePath string func now() time.Time { return time.Now().UTC() } // --------------------------------------------------------------------------------------------- // Handlers // --------------------------------------------------------------------------------------------- func createEASRegistration(w http.ResponseWriter, r *http.Request) { reg, ok := readRegistrationBody(w, r) if !ok { return } if errs := validateEasRegistration(reg, now()); errs != nil { problem(w, http.StatusBadRequest, "Invalid EAS registration", "", errs) return } easMutex.Lock() defer easMutex.Unlock() if _, err := GetEasRegistrationIdByEasId(reg.EasProf.EasId); err == nil { problem(w, http.StatusForbidden, "EAS "+reg.EasProf.EasId+" is already registered on this EES; update the existing registration", causeEasAlreadyRegistered, nil) return } links, err := registerEasWithMec011(r.Context(), reg.EasProf) if err != nil { log.Error("EAS registration: ", err.Error()) problem(w, http.StatusInternalServerError, "Failed to register the EAS with the MEC platform", "", nil) return } expiry := grantedExpiry(reg.ExpTime, now()) reg.ExpTime = &expiry registrationId := uuid.New().String() rec := &easRecord{Registration: *reg, Links: links} if err := saveRecord(registrationId, rec); err != nil { log.Error("EAS registration: ", err.Error()) _ = deregisterEasFromMec011(context.Background(), links) problem(w, http.StatusInternalServerError, "Failed to store the EAS registration", "", nil) return } scheduleExpiry(registrationId, expiry) log.Info("EAS ", reg.EasProf.EasId, " registered: registrationId ", registrationId, ", app instance ", links.AppInstanceId) w.Header().Set("Location", registrationUri(registrationId)) writeJson(w, http.StatusCreated, reg) } func readIndEASRegistration(w http.ResponseWriter, r *http.Request) { rec, err := loadRecord(mux.Vars(r)["registrationId"]) if err != nil { notFoundOrError(w, err) return } writeJson(w, http.StatusOK, rec.Registration) } func updateIndEASRegistration(w http.ResponseWriter, r *http.Request) { reg, ok := readRegistrationBody(w, r) if !ok { return } if reg.ExpTime == nil { // Full replacement: an absent expiry time means the default lifetime from now expiry := grantedExpiry(nil, now()) reg.ExpTime = &expiry } applyUpdate(w, r, mux.Vars(r)["registrationId"], func(*easRecord) (*EasRegistration, []InvalidParam) { return reg, nil }) } func modifyIndEASRegistration(w http.ResponseWriter, r *http.Request) { patch, err := ioutil.ReadAll(r.Body) if err != nil || !json.Valid(patch) { problem(w, http.StatusBadRequest, "Invalid JSON merge patch", "", nil) return } applyUpdate(w, r, mux.Vars(r)["registrationId"], func(rec *easRecord) (*EasRegistration, []InvalidParam) { // JSON Merge Patch (RFC 7396) applied to the stored registration (implementation constraint I1) patched, err := applyMergePatch(rec.Registration, patch) if err != nil { return nil, []InvalidParam{{Param: "/", Reason: err.Error()}} } return patched, nil }) } func deleteIndEASRegistration(w http.ResponseWriter, r *http.Request) { registrationId := mux.Vars(r)["registrationId"] easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil { notFoundOrError(w, err) return } if err := deregister(registrationId, rec, false); err != nil { problem(w, http.StatusInternalServerError, "Failed to deregister the EAS from the MEC platform; retry the deletion", "", nil) return } w.WriteHeader(http.StatusNoContent) } // applyUpdate - common part of PUT and PATCH: the new registration is validated, mirrored into the // MEC platform and stored, and its expiry timer is restarted. func applyUpdate(w http.ResponseWriter, r *http.Request, registrationId string, newRegistration func(*easRecord) (*EasRegistration, []InvalidParam)) { easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil { notFoundOrError(w, err) return } reg, errs := newRegistration(rec) if errs == nil { errs = validateEasRegistration(reg, now()) } if errs == nil && reg.EasProf.EasId != rec.Registration.EasProf.EasId { errs = []InvalidParam{{Param: "/easProf/easId", Reason: "the EAS identifier of a registration cannot be changed"}} } if errs != nil { problem(w, http.StatusBadRequest, "Invalid EAS registration update", "", errs) return } links, err := updateEasInMec011(r.Context(), rec.Links, reg.EasProf) rec.Links = links if err != nil { log.Error("EAS registration update: ", err.Error()) _ = saveRecord(registrationId, rec) // keep the resource ids that still exist problem(w, http.StatusInternalServerError, "Failed to update the EAS in the MEC platform", "", nil) return } expiry := grantedExpiry(reg.ExpTime, now()) reg.ExpTime = &expiry rec.Registration = *reg if err := saveRecord(registrationId, rec); err != nil { problem(w, http.StatusInternalServerError, "Failed to store the EAS registration", "", nil) return } scheduleExpiry(registrationId, expiry) writeJson(w, http.StatusOK, reg) } // --------------------------------------------------------------------------------------------- // Deregistration, expiry, reconciliation, platform events // --------------------------------------------------------------------------------------------- // deregister - remove an EAS registration everywhere (caller holds easMutex). If the MEC platform // clean-up fails, the record is kept unless force is set, so that an explicit DELETE can be retried // (missing MEC platform resources are ignored, so a retry is idempotent). Expiry, app instance removal // and flush use force: nobody could retry them, leftovers are logged. func deregister(registrationId string, rec *easRecord, force bool) error { mecErr := deregisterEasFromMec011(context.Background(), rec.Links) if mecErr != nil && !force { return mecErr } stopExpiry(registrationId) if err := deleteEasRegistration(registrationId, rec.Registration.EasProf.EasId); err != nil && err != ErrNotFound { return err } log.Info("EAS ", rec.Registration.EasProf.EasId, " deregistered: registrationId ", registrationId) return mecErr } func scheduleExpiry(registrationId string, expiry time.Time) { stopExpiry(registrationId) expiryTimers[registrationId] = time.AfterFunc(time.Until(expiry), func() { expire(registrationId) }) } func stopExpiry(registrationId string) { if t, found := expiryTimers[registrationId]; found { t.Stop() delete(expiryTimers, registrationId) } } func expire(registrationId string) { easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil || rec.Registration.ExpTime == nil || rec.Registration.ExpTime.After(now()) { return // deleted or renewed in the meantime } log.Info("EAS registration ", registrationId, " expired") _ = deregister(registrationId, rec, true) } // reconcile - after a restart: expired registrations and registrations whose app instance no longer // exists are removed, the other ones get their expiry timer back. Retried until the MEC platform answers. func reconcile() { deadline := time.Now().Add(reconcileTimeout) for { err := reconcileOnce() if err == nil || time.Now().After(deadline) { if err != nil { log.Error("EAS registration reconciliation incomplete: ", err.Error()) } return } time.Sleep(reconcileRetryInterval) } } func reconcileOnce() error { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return err } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) != nil || rec.Registration.EasProf == nil { log.Error("Dropping unreadable EAS registration record ", registrationId) _ = deleteEasRegistration(registrationId, "") continue } exists, err := appInstanceExists(context.Background(), rec.Links.AppInstanceId) if err != nil { return err } switch { case !exists: log.Info("EAS registration ", registrationId, ": app instance gone, removing the registration") _ = deregister(registrationId, &rec, true) case rec.Registration.ExpTime == nil || !rec.Registration.ExpTime.After(now()): log.Info("EAS registration ", registrationId, " expired while the service was down") _ = deregister(registrationId, &rec, true) default: scheduleExpiry(registrationId, *rec.Registration.ExpTime) } } return nil } // msgHandler - platform events. Returns immediately: the MQ listener is shared with the MEC 011 // services, which must keep receiving events while an EAS operation is in progress. func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgAppRemove: appInstanceId := msg.Payload["id"] go onAppRemoved(appInstanceId) case mq.MsgAppFlush: go onAppsFlushed() default: } } // onAppRemoved - the app instance of an EAS was removed outside EDGE-3 (e.g. sandbox-ctrl API): // the EAS registration and its MEC 011 registration are removed too. func onAppRemoved(appInstanceId string) { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) == nil && rec.Links.AppInstanceId == appInstanceId && rec.Registration.EasProf != nil { log.Info("App instance ", appInstanceId, " of EAS registration ", registrationId, " removed by the platform") _ = deregister(registrationId, &rec, true) } } } // onAppsFlushed - scenario terminated or sandbox reset: all EAS registrations of this MEP are removed func onAppsFlushed() { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) == nil && rec.Registration.EasProf != nil { _ = deregister(registrationId, &rec, true) } } for registrationId := range expiryTimers { stopExpiry(registrationId) } _ = flushEasRegistrations() log.Info("EAS registrations flushed") } // --------------------------------------------------------------------------------------------- // helpers // --------------------------------------------------------------------------------------------- func loadRecord(registrationId string) (*easRecord, error) { value, err := GetEasRegistration(registrationId) if err != nil { return nil, err } var rec easRecord if err := json.Unmarshal([]byte(value), &rec); err != nil || rec.Registration.EasProf == nil { return nil, errors.New("unreadable EAS registration record " + registrationId) } return &rec, nil } func saveRecord(registrationId string, rec *easRecord) error { value, err := json.Marshal(rec) if err != nil { return err } return setEasRegistration(registrationId, rec.Registration.EasProf.EasId, string(value)) } func readRegistrationBody(w http.ResponseWriter, r *http.Request) (*EasRegistration, bool) { var reg EasRegistration if r.Body == nil || json.NewDecoder(r.Body).Decode(®) != nil { problem(w, http.StatusBadRequest, "Invalid JSON body", "", nil) return nil, false } return ®, true } // applyMergePatch - JSON Merge Patch (RFC 7396) of an EAS registration func applyMergePatch(reg EasRegistration, patch []byte) (*EasRegistration, error) { var target, p interface{} original, _ := json.Marshal(reg) _ = json.Unmarshal(original, &target) if err := json.Unmarshal(patch, &p); err != nil { return nil, err } merged, _ := json.Marshal(mergePatch(target, p)) var out EasRegistration if err := json.Unmarshal(merged, &out); err != nil { return nil, errors.New("patched registration is not a valid EASRegistration: " + err.Error()) } return &out, nil } func mergePatch(target interface{}, patch interface{}) interface{} { p, ok := patch.(map[string]interface{}) if !ok { return patch } t, ok := target.(map[string]interface{}) if !ok { t = map[string]interface{}{} } for k, v := range p { if v == nil { delete(t, k) } else { t[k] = mergePatch(t[k], v) } } return t } func registrationUri(registrationId string) string { host := "" if hostUrl != nil { host = strings.TrimSuffix(hostUrl.String(), "/") } return host + basePath + "registrations/" + registrationId } func notFoundOrError(w http.ResponseWriter, err error) { if err == ErrNotFound { problem(w, http.StatusNotFound, "EAS registration not found", "", nil) return } log.Error(err.Error()) problem(w, http.StatusInternalServerError, "Failed to read the EAS registration", "", nil) } func problem(w http.ResponseWriter, code int, detail string, cause string, invalid []InvalidParam) { status := int32(code) pd := ProblemDetails{Title: http.StatusText(code), Status: &status, Detail: detail, Cause: cause, InvalidParams: invalid} w.Header().Set("Content-Type", "application/problem+json") w.WriteHeader(code) _ = json.NewEncoder(w).Encode(pd) } func writeJson(w http.ResponseWriter, code int, v interface{}) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) _ = json.NewEncoder(w).Encode(v) } func asJsonValue(v interface{}) interface{} { var out interface{} b, _ := json.Marshal(v) _ = json.Unmarshal(b, &out) return out } go-apps/meep-app-enablement/server/edge-3/eas_lifecycle_test.go 0 → 100644 +670 −0 File added.Preview size limit exceeded, changes collapsed. Show changes go-apps/meep-app-enablement/server/edge-3/edge3.go +37 −48 Changes for go-apps/meep-app-enablement/server/edge-3/edge3.go: 37 added lines, 48 removed lines. Original line number Diff line number Diff line Loading @@ -19,8 +19,6 @@ package server import ( "encoding/json" "net/http" "net/url" "sync" Loading @@ -29,16 +27,25 @@ import ( ) const serviceName = "EDGE-3 Service" const easRegistrationBasePath = "eees-easregistration/v1/" // In-cluster services used by the adapter (same addressing as the platform MEC services) const sandboxCtrlUrl = "http://meep-sandbox-ctrl" var sandboxName string var mepName string var hostUrl *url.URL var mqLocal *mq.MsgQueue var handlerId int // Init - EDGE-3 service initialization func Init(sandbox string, mep string, host *url.URL, msgQueue *mq.MsgQueue, redisAddr_ string, globalMutex *sync.Mutex) (err error) { // Same signature as the other app-enablement services; the host URL, message queue and global // mutex are used by the connector from the next phases onwards. // The app-enablement global mutex is deliberately not used: EAS operations call the MEC 011 // handlers of this process, which take it (see easMutex). sandboxName = sandbox mepName = mep hostUrl = host mqLocal = msgQueue // Connect the EAS registration store owned by this service err = initStore(sandbox, mep, redisAddr_) Loading @@ -46,61 +53,43 @@ func Init(sandbox string, mep string, host *url.URL, msgQueue *mq.MsgQueue, redi log.Error("Failed to connect to Redis DB. Error: ", err) return err } // Resource URIs and MEC 011 adapter clients (MEP in-cluster service, as for the platform MEC services) appEnablementUrl := "http://" + mep + "-meep-app-enablement" if mep == globalLocationName { basePath = "/" + sandbox + "/" + easRegistrationBasePath appEnablementUrl = "http://meep-app-enablement" } else { basePath = "/" + sandbox + "/" + mep + "/" + easRegistrationBasePath } initMec011Clients(sandboxCtrlUrl, appEnablementUrl) log.Info(serviceName, " initialized: sandbox ", sandboxName, ", MEP ", mepName) return nil } // Run - Start EDGE-3 service func Run() (err error) { // Platform events (app instance removal, flush) handlerId, err = mqLocal.RegisterHandler(mq.MsgHandler{Handler: msgHandler, UserData: nil}) if err != nil { log.Error("Failed to register MQ handler: ", err.Error()) return err } // Stored registrations: expiry timers and clean-up, once the MEC platform answers go reconcile() return nil } // Stop - Stop EDGE-3 service func Stop() (err error) { return nil } // Eees_EASRegistration (EDGE-3) func createEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "CreateEASRegistration") if mqLocal != nil { mqLocal.UnregisterHandler(handlerId) } func readIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "ReadIndEASRegistration") easMutex.Lock() defer easMutex.Unlock() for registrationId := range expiryTimers { stopExpiry(registrationId) } func updateIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "UpdateIndEASRegistration") } func modifyIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "ModifyIndEASRegistration") } func deleteIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "DeleteIndEASRegistration") } // notImplemented - operation supported by the sandbox but not implemented yet func notImplemented(w http.ResponseWriter, operation string) { errHandlerProblemDetails(w, "Not Implemented", operation+" is not implemented yet", http.StatusNotImplemented) } func errHandlerProblemDetails(w http.ResponseWriter, title string, detail string, code int) { status := int32(code) pd := ProblemDetails{ Title: title, Status: &status, Detail: detail, } jsonResponse, err := json.Marshal(pd) if err != nil { log.Error(err.Error()) w.WriteHeader(http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/problem+json") w.WriteHeader(code) _, _ = w.Write(jsonResponse) return nil } go-apps/meep-app-enablement/server/edge-3/mec011_adapter.go +5 −2 Changes for go-apps/meep-app-enablement/server/edge-3/mec011_adapter.go: 5 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -159,11 +159,14 @@ func toServiceInfo(prof *EasProfile) (smc.ServiceInfo, error) { Protocol: protocol, Version: "2.0", Endpoint: endpoint, Security: &smc.SecurityInfo{}, }, Serializer: &serializer, ScopeOfLocality: &locality, // Consumed by application clients on UEs, not only by MEC applications of the same locality ConsumedLocalOnly: false, // MEC 011 service-mgmt defaults an omitted consumedLocalOnly to true, and the generated client // omits false: state the effective value (it concerns MEC application consumers only; the // application clients on UEs reach the EAS through EDGE-1 EAS discovery, not Mp1). ConsumedLocalOnly: true, }, nil } Loading go-apps/meep-app-enablement/server/edge-3/mec011_adapter_test.go +26 −1 Changes for go-apps/meep-app-enablement/server/edge-3/mec011_adapter_test.go: 26 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -248,7 +248,8 @@ func TestToServiceInfo(t *testing.T) { t.Fatal(err) } if svc.SerName != "eas-v2x-1" || svc.SerCategory.Id != "EAS" || svc.Version == "" || *svc.State != smc.ACTIVE_ServiceState || *svc.Serializer != smc.JSON_SerializerType || *svc.ScopeOfLocality != smc.MEC_HOST_LocalityType || svc.ConsumedLocalOnly { *svc.Serializer != smc.JSON_SerializerType || *svc.ScopeOfLocality != smc.MEC_HOST_LocalityType || !svc.ConsumedLocalOnly || svc.TransportInfo.Security == nil { t.Errorf("service info: %s", asJson(t, svc)) } ti := svc.TransportInfo Loading Loading @@ -283,3 +284,27 @@ func containsAll(s string, parts ...string) bool { } return true } // The in-place/re-publish decision compares the service as MEC 011 service-mgmt returns it (platform // attributes and defaults set) with the adapter ServiceInfo. func TestSameServiceDefinition(t *testing.T) { prof := decodeProfile(t, fullEasProfile) desired, _ := toServiceInfo(prof) stored := `{"serInstanceId":"svc-1","serName":"eas-v2x-1","serCategory":{"href":"` + easCategoryHref + `","id":"EAS","name":"EAS","version":"19.7.0"}, "version":"1.0","state":"INACTIVE","transportInfo":{"id":"eas-endpoint","name":"EAS endpoint","type":"REST_HTTP","protocol":"HTTP","version":"2.0", "endpoint":{"uris":["https://eas-v2x-1.mep2.example/api"],"fqdn":null,"addresses":null,"alternative":null},"security":{}}, "serializer":"JSON","scopeOfLocality":"MEC_HOST","consumedLocalOnly":true,"isLocal":true, "_links":{"self":{"href":"https://sandbox.example/s/mep2/mec_service_mgmt/v1/applications/a/services/svc-1"}}}` var current smc.ServiceInfo if err := json.Unmarshal([]byte(stored), ¤t); err != nil { t.Fatal(err) } if !sameServiceDefinition(¤t, &desired) { t.Errorf("same definition (state and platform attributes differ only) not recognised:\n%s\n%s", asJson(t, current), asJson(t, desired)) } moved := decodeProfile(t, strings.Replace(fullEasProfile, "eas-v2x-1.mep2.example", "elsewhere.example", 1)) other, _ := toServiceInfo(moved) if sameServiceDefinition(¤t, &other) { t.Errorf("endpoint change not detected") } } Loading
go-apps/meep-app-enablement/server/edge-3/eas_lifecycle.go 0 → 100644 +457 −0 Changes for go-apps/meep-app-enablement/server/edge-3/eas_lifecycle.go: 457 added lines, 0 removed lines. Original line number Diff line number Diff line /* * Copyright (c) 2026 The AdvantEDGE Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // EAS registration lifecycle (3GPP TS 29.558 Eees_EASRegistration): create, read, update, modify, // delete and expiry of the EAS registrations of this MEP. // // Every registration is mirrored into the MEC platform through the adapter (app instance, MEC 011 // registration and service), so that native MEC applications see the EAS over Mp1. The registration // record (EAS registration as granted + the MEC platform resource ids) is kept in the EDGE-3 store, // which also makes the expiry timers restart-safe. package server import ( "context" "encoding/json" "errors" "io/ioutil" "net/http" "strings" "sync" "time" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" "github.com/google/uuid" "github.com/gorilla/mux" ) // ProblemDetails cause of a second registration of the same EAS on this MEP (sandbox-defined) const causeEasAlreadyRegistered = "EAS_ALREADY_REGISTERED" // Reconciliation of stored registrations after a restart: retried until the MEC platform answers var reconcileRetryInterval = 5 * time.Second var reconcileTimeout = 5 * time.Minute // easRecord - EAS registration record of the EDGE-3 store type easRecord struct { Registration EasRegistration `json:"registration"` Links easLinks `json:"links"` } // easMutex serializes the EAS registration operations of this MEP. It is never held while waiting for // the app-enablement global mutex (the MEC 011 handlers take that one). var easMutex sync.Mutex var expiryTimers = map[string]*time.Timer{} var basePath string func now() time.Time { return time.Now().UTC() } // --------------------------------------------------------------------------------------------- // Handlers // --------------------------------------------------------------------------------------------- func createEASRegistration(w http.ResponseWriter, r *http.Request) { reg, ok := readRegistrationBody(w, r) if !ok { return } if errs := validateEasRegistration(reg, now()); errs != nil { problem(w, http.StatusBadRequest, "Invalid EAS registration", "", errs) return } easMutex.Lock() defer easMutex.Unlock() if _, err := GetEasRegistrationIdByEasId(reg.EasProf.EasId); err == nil { problem(w, http.StatusForbidden, "EAS "+reg.EasProf.EasId+" is already registered on this EES; update the existing registration", causeEasAlreadyRegistered, nil) return } links, err := registerEasWithMec011(r.Context(), reg.EasProf) if err != nil { log.Error("EAS registration: ", err.Error()) problem(w, http.StatusInternalServerError, "Failed to register the EAS with the MEC platform", "", nil) return } expiry := grantedExpiry(reg.ExpTime, now()) reg.ExpTime = &expiry registrationId := uuid.New().String() rec := &easRecord{Registration: *reg, Links: links} if err := saveRecord(registrationId, rec); err != nil { log.Error("EAS registration: ", err.Error()) _ = deregisterEasFromMec011(context.Background(), links) problem(w, http.StatusInternalServerError, "Failed to store the EAS registration", "", nil) return } scheduleExpiry(registrationId, expiry) log.Info("EAS ", reg.EasProf.EasId, " registered: registrationId ", registrationId, ", app instance ", links.AppInstanceId) w.Header().Set("Location", registrationUri(registrationId)) writeJson(w, http.StatusCreated, reg) } func readIndEASRegistration(w http.ResponseWriter, r *http.Request) { rec, err := loadRecord(mux.Vars(r)["registrationId"]) if err != nil { notFoundOrError(w, err) return } writeJson(w, http.StatusOK, rec.Registration) } func updateIndEASRegistration(w http.ResponseWriter, r *http.Request) { reg, ok := readRegistrationBody(w, r) if !ok { return } if reg.ExpTime == nil { // Full replacement: an absent expiry time means the default lifetime from now expiry := grantedExpiry(nil, now()) reg.ExpTime = &expiry } applyUpdate(w, r, mux.Vars(r)["registrationId"], func(*easRecord) (*EasRegistration, []InvalidParam) { return reg, nil }) } func modifyIndEASRegistration(w http.ResponseWriter, r *http.Request) { patch, err := ioutil.ReadAll(r.Body) if err != nil || !json.Valid(patch) { problem(w, http.StatusBadRequest, "Invalid JSON merge patch", "", nil) return } applyUpdate(w, r, mux.Vars(r)["registrationId"], func(rec *easRecord) (*EasRegistration, []InvalidParam) { // JSON Merge Patch (RFC 7396) applied to the stored registration (implementation constraint I1) patched, err := applyMergePatch(rec.Registration, patch) if err != nil { return nil, []InvalidParam{{Param: "/", Reason: err.Error()}} } return patched, nil }) } func deleteIndEASRegistration(w http.ResponseWriter, r *http.Request) { registrationId := mux.Vars(r)["registrationId"] easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil { notFoundOrError(w, err) return } if err := deregister(registrationId, rec, false); err != nil { problem(w, http.StatusInternalServerError, "Failed to deregister the EAS from the MEC platform; retry the deletion", "", nil) return } w.WriteHeader(http.StatusNoContent) } // applyUpdate - common part of PUT and PATCH: the new registration is validated, mirrored into the // MEC platform and stored, and its expiry timer is restarted. func applyUpdate(w http.ResponseWriter, r *http.Request, registrationId string, newRegistration func(*easRecord) (*EasRegistration, []InvalidParam)) { easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil { notFoundOrError(w, err) return } reg, errs := newRegistration(rec) if errs == nil { errs = validateEasRegistration(reg, now()) } if errs == nil && reg.EasProf.EasId != rec.Registration.EasProf.EasId { errs = []InvalidParam{{Param: "/easProf/easId", Reason: "the EAS identifier of a registration cannot be changed"}} } if errs != nil { problem(w, http.StatusBadRequest, "Invalid EAS registration update", "", errs) return } links, err := updateEasInMec011(r.Context(), rec.Links, reg.EasProf) rec.Links = links if err != nil { log.Error("EAS registration update: ", err.Error()) _ = saveRecord(registrationId, rec) // keep the resource ids that still exist problem(w, http.StatusInternalServerError, "Failed to update the EAS in the MEC platform", "", nil) return } expiry := grantedExpiry(reg.ExpTime, now()) reg.ExpTime = &expiry rec.Registration = *reg if err := saveRecord(registrationId, rec); err != nil { problem(w, http.StatusInternalServerError, "Failed to store the EAS registration", "", nil) return } scheduleExpiry(registrationId, expiry) writeJson(w, http.StatusOK, reg) } // --------------------------------------------------------------------------------------------- // Deregistration, expiry, reconciliation, platform events // --------------------------------------------------------------------------------------------- // deregister - remove an EAS registration everywhere (caller holds easMutex). If the MEC platform // clean-up fails, the record is kept unless force is set, so that an explicit DELETE can be retried // (missing MEC platform resources are ignored, so a retry is idempotent). Expiry, app instance removal // and flush use force: nobody could retry them, leftovers are logged. func deregister(registrationId string, rec *easRecord, force bool) error { mecErr := deregisterEasFromMec011(context.Background(), rec.Links) if mecErr != nil && !force { return mecErr } stopExpiry(registrationId) if err := deleteEasRegistration(registrationId, rec.Registration.EasProf.EasId); err != nil && err != ErrNotFound { return err } log.Info("EAS ", rec.Registration.EasProf.EasId, " deregistered: registrationId ", registrationId) return mecErr } func scheduleExpiry(registrationId string, expiry time.Time) { stopExpiry(registrationId) expiryTimers[registrationId] = time.AfterFunc(time.Until(expiry), func() { expire(registrationId) }) } func stopExpiry(registrationId string) { if t, found := expiryTimers[registrationId]; found { t.Stop() delete(expiryTimers, registrationId) } } func expire(registrationId string) { easMutex.Lock() defer easMutex.Unlock() rec, err := loadRecord(registrationId) if err != nil || rec.Registration.ExpTime == nil || rec.Registration.ExpTime.After(now()) { return // deleted or renewed in the meantime } log.Info("EAS registration ", registrationId, " expired") _ = deregister(registrationId, rec, true) } // reconcile - after a restart: expired registrations and registrations whose app instance no longer // exists are removed, the other ones get their expiry timer back. Retried until the MEC platform answers. func reconcile() { deadline := time.Now().Add(reconcileTimeout) for { err := reconcileOnce() if err == nil || time.Now().After(deadline) { if err != nil { log.Error("EAS registration reconciliation incomplete: ", err.Error()) } return } time.Sleep(reconcileRetryInterval) } } func reconcileOnce() error { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return err } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) != nil || rec.Registration.EasProf == nil { log.Error("Dropping unreadable EAS registration record ", registrationId) _ = deleteEasRegistration(registrationId, "") continue } exists, err := appInstanceExists(context.Background(), rec.Links.AppInstanceId) if err != nil { return err } switch { case !exists: log.Info("EAS registration ", registrationId, ": app instance gone, removing the registration") _ = deregister(registrationId, &rec, true) case rec.Registration.ExpTime == nil || !rec.Registration.ExpTime.After(now()): log.Info("EAS registration ", registrationId, " expired while the service was down") _ = deregister(registrationId, &rec, true) default: scheduleExpiry(registrationId, *rec.Registration.ExpTime) } } return nil } // msgHandler - platform events. Returns immediately: the MQ listener is shared with the MEC 011 // services, which must keep receiving events while an EAS operation is in progress. func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgAppRemove: appInstanceId := msg.Payload["id"] go onAppRemoved(appInstanceId) case mq.MsgAppFlush: go onAppsFlushed() default: } } // onAppRemoved - the app instance of an EAS was removed outside EDGE-3 (e.g. sandbox-ctrl API): // the EAS registration and its MEC 011 registration are removed too. func onAppRemoved(appInstanceId string) { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) == nil && rec.Links.AppInstanceId == appInstanceId && rec.Registration.EasProf != nil { log.Info("App instance ", appInstanceId, " of EAS registration ", registrationId, " removed by the platform") _ = deregister(registrationId, &rec, true) } } } // onAppsFlushed - scenario terminated or sandbox reset: all EAS registrations of this MEP are removed func onAppsFlushed() { easMutex.Lock() defer easMutex.Unlock() registrations, err := ListEasRegistrations() if err != nil { return } for registrationId, value := range registrations { var rec easRecord if json.Unmarshal([]byte(value), &rec) == nil && rec.Registration.EasProf != nil { _ = deregister(registrationId, &rec, true) } } for registrationId := range expiryTimers { stopExpiry(registrationId) } _ = flushEasRegistrations() log.Info("EAS registrations flushed") } // --------------------------------------------------------------------------------------------- // helpers // --------------------------------------------------------------------------------------------- func loadRecord(registrationId string) (*easRecord, error) { value, err := GetEasRegistration(registrationId) if err != nil { return nil, err } var rec easRecord if err := json.Unmarshal([]byte(value), &rec); err != nil || rec.Registration.EasProf == nil { return nil, errors.New("unreadable EAS registration record " + registrationId) } return &rec, nil } func saveRecord(registrationId string, rec *easRecord) error { value, err := json.Marshal(rec) if err != nil { return err } return setEasRegistration(registrationId, rec.Registration.EasProf.EasId, string(value)) } func readRegistrationBody(w http.ResponseWriter, r *http.Request) (*EasRegistration, bool) { var reg EasRegistration if r.Body == nil || json.NewDecoder(r.Body).Decode(®) != nil { problem(w, http.StatusBadRequest, "Invalid JSON body", "", nil) return nil, false } return ®, true } // applyMergePatch - JSON Merge Patch (RFC 7396) of an EAS registration func applyMergePatch(reg EasRegistration, patch []byte) (*EasRegistration, error) { var target, p interface{} original, _ := json.Marshal(reg) _ = json.Unmarshal(original, &target) if err := json.Unmarshal(patch, &p); err != nil { return nil, err } merged, _ := json.Marshal(mergePatch(target, p)) var out EasRegistration if err := json.Unmarshal(merged, &out); err != nil { return nil, errors.New("patched registration is not a valid EASRegistration: " + err.Error()) } return &out, nil } func mergePatch(target interface{}, patch interface{}) interface{} { p, ok := patch.(map[string]interface{}) if !ok { return patch } t, ok := target.(map[string]interface{}) if !ok { t = map[string]interface{}{} } for k, v := range p { if v == nil { delete(t, k) } else { t[k] = mergePatch(t[k], v) } } return t } func registrationUri(registrationId string) string { host := "" if hostUrl != nil { host = strings.TrimSuffix(hostUrl.String(), "/") } return host + basePath + "registrations/" + registrationId } func notFoundOrError(w http.ResponseWriter, err error) { if err == ErrNotFound { problem(w, http.StatusNotFound, "EAS registration not found", "", nil) return } log.Error(err.Error()) problem(w, http.StatusInternalServerError, "Failed to read the EAS registration", "", nil) } func problem(w http.ResponseWriter, code int, detail string, cause string, invalid []InvalidParam) { status := int32(code) pd := ProblemDetails{Title: http.StatusText(code), Status: &status, Detail: detail, Cause: cause, InvalidParams: invalid} w.Header().Set("Content-Type", "application/problem+json") w.WriteHeader(code) _ = json.NewEncoder(w).Encode(pd) } func writeJson(w http.ResponseWriter, code int, v interface{}) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) _ = json.NewEncoder(w).Encode(v) } func asJsonValue(v interface{}) interface{} { var out interface{} b, _ := json.Marshal(v) _ = json.Unmarshal(b, &out) return out }
go-apps/meep-app-enablement/server/edge-3/eas_lifecycle_test.go 0 → 100644 +670 −0 File added.Preview size limit exceeded, changes collapsed. Show changes
go-apps/meep-app-enablement/server/edge-3/edge3.go +37 −48 Changes for go-apps/meep-app-enablement/server/edge-3/edge3.go: 37 added lines, 48 removed lines. Original line number Diff line number Diff line Loading @@ -19,8 +19,6 @@ package server import ( "encoding/json" "net/http" "net/url" "sync" Loading @@ -29,16 +27,25 @@ import ( ) const serviceName = "EDGE-3 Service" const easRegistrationBasePath = "eees-easregistration/v1/" // In-cluster services used by the adapter (same addressing as the platform MEC services) const sandboxCtrlUrl = "http://meep-sandbox-ctrl" var sandboxName string var mepName string var hostUrl *url.URL var mqLocal *mq.MsgQueue var handlerId int // Init - EDGE-3 service initialization func Init(sandbox string, mep string, host *url.URL, msgQueue *mq.MsgQueue, redisAddr_ string, globalMutex *sync.Mutex) (err error) { // Same signature as the other app-enablement services; the host URL, message queue and global // mutex are used by the connector from the next phases onwards. // The app-enablement global mutex is deliberately not used: EAS operations call the MEC 011 // handlers of this process, which take it (see easMutex). sandboxName = sandbox mepName = mep hostUrl = host mqLocal = msgQueue // Connect the EAS registration store owned by this service err = initStore(sandbox, mep, redisAddr_) Loading @@ -46,61 +53,43 @@ func Init(sandbox string, mep string, host *url.URL, msgQueue *mq.MsgQueue, redi log.Error("Failed to connect to Redis DB. Error: ", err) return err } // Resource URIs and MEC 011 adapter clients (MEP in-cluster service, as for the platform MEC services) appEnablementUrl := "http://" + mep + "-meep-app-enablement" if mep == globalLocationName { basePath = "/" + sandbox + "/" + easRegistrationBasePath appEnablementUrl = "http://meep-app-enablement" } else { basePath = "/" + sandbox + "/" + mep + "/" + easRegistrationBasePath } initMec011Clients(sandboxCtrlUrl, appEnablementUrl) log.Info(serviceName, " initialized: sandbox ", sandboxName, ", MEP ", mepName) return nil } // Run - Start EDGE-3 service func Run() (err error) { // Platform events (app instance removal, flush) handlerId, err = mqLocal.RegisterHandler(mq.MsgHandler{Handler: msgHandler, UserData: nil}) if err != nil { log.Error("Failed to register MQ handler: ", err.Error()) return err } // Stored registrations: expiry timers and clean-up, once the MEC platform answers go reconcile() return nil } // Stop - Stop EDGE-3 service func Stop() (err error) { return nil } // Eees_EASRegistration (EDGE-3) func createEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "CreateEASRegistration") if mqLocal != nil { mqLocal.UnregisterHandler(handlerId) } func readIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "ReadIndEASRegistration") easMutex.Lock() defer easMutex.Unlock() for registrationId := range expiryTimers { stopExpiry(registrationId) } func updateIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "UpdateIndEASRegistration") } func modifyIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "ModifyIndEASRegistration") } func deleteIndEASRegistration(w http.ResponseWriter, r *http.Request) { notImplemented(w, "DeleteIndEASRegistration") } // notImplemented - operation supported by the sandbox but not implemented yet func notImplemented(w http.ResponseWriter, operation string) { errHandlerProblemDetails(w, "Not Implemented", operation+" is not implemented yet", http.StatusNotImplemented) } func errHandlerProblemDetails(w http.ResponseWriter, title string, detail string, code int) { status := int32(code) pd := ProblemDetails{ Title: title, Status: &status, Detail: detail, } jsonResponse, err := json.Marshal(pd) if err != nil { log.Error(err.Error()) w.WriteHeader(http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/problem+json") w.WriteHeader(code) _, _ = w.Write(jsonResponse) return nil }
go-apps/meep-app-enablement/server/edge-3/mec011_adapter.go +5 −2 Changes for go-apps/meep-app-enablement/server/edge-3/mec011_adapter.go: 5 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -159,11 +159,14 @@ func toServiceInfo(prof *EasProfile) (smc.ServiceInfo, error) { Protocol: protocol, Version: "2.0", Endpoint: endpoint, Security: &smc.SecurityInfo{}, }, Serializer: &serializer, ScopeOfLocality: &locality, // Consumed by application clients on UEs, not only by MEC applications of the same locality ConsumedLocalOnly: false, // MEC 011 service-mgmt defaults an omitted consumedLocalOnly to true, and the generated client // omits false: state the effective value (it concerns MEC application consumers only; the // application clients on UEs reach the EAS through EDGE-1 EAS discovery, not Mp1). ConsumedLocalOnly: true, }, nil } Loading
go-apps/meep-app-enablement/server/edge-3/mec011_adapter_test.go +26 −1 Changes for go-apps/meep-app-enablement/server/edge-3/mec011_adapter_test.go: 26 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -248,7 +248,8 @@ func TestToServiceInfo(t *testing.T) { t.Fatal(err) } if svc.SerName != "eas-v2x-1" || svc.SerCategory.Id != "EAS" || svc.Version == "" || *svc.State != smc.ACTIVE_ServiceState || *svc.Serializer != smc.JSON_SerializerType || *svc.ScopeOfLocality != smc.MEC_HOST_LocalityType || svc.ConsumedLocalOnly { *svc.Serializer != smc.JSON_SerializerType || *svc.ScopeOfLocality != smc.MEC_HOST_LocalityType || !svc.ConsumedLocalOnly || svc.TransportInfo.Security == nil { t.Errorf("service info: %s", asJson(t, svc)) } ti := svc.TransportInfo Loading Loading @@ -283,3 +284,27 @@ func containsAll(s string, parts ...string) bool { } return true } // The in-place/re-publish decision compares the service as MEC 011 service-mgmt returns it (platform // attributes and defaults set) with the adapter ServiceInfo. func TestSameServiceDefinition(t *testing.T) { prof := decodeProfile(t, fullEasProfile) desired, _ := toServiceInfo(prof) stored := `{"serInstanceId":"svc-1","serName":"eas-v2x-1","serCategory":{"href":"` + easCategoryHref + `","id":"EAS","name":"EAS","version":"19.7.0"}, "version":"1.0","state":"INACTIVE","transportInfo":{"id":"eas-endpoint","name":"EAS endpoint","type":"REST_HTTP","protocol":"HTTP","version":"2.0", "endpoint":{"uris":["https://eas-v2x-1.mep2.example/api"],"fqdn":null,"addresses":null,"alternative":null},"security":{}}, "serializer":"JSON","scopeOfLocality":"MEC_HOST","consumedLocalOnly":true,"isLocal":true, "_links":{"self":{"href":"https://sandbox.example/s/mep2/mec_service_mgmt/v1/applications/a/services/svc-1"}}}` var current smc.ServiceInfo if err := json.Unmarshal([]byte(stored), ¤t); err != nil { t.Fatal(err) } if !sameServiceDefinition(¤t, &desired) { t.Errorf("same definition (state and platform attributes differ only) not recognised:\n%s\n%s", asJson(t, current), asJson(t, desired)) } moved := decodeProfile(t, strings.Replace(fullEasProfile, "eas-v2x-1.mep2.example", "elsewhere.example", 1)) other, _ := toServiceInfo(moved) if sameServiceDefinition(¤t, &other) { t.Errorf("endpoint change not detected") } }