Validate federation_test scripts (c3b0f335) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-apps/meep-federation/go.mod

+1 −0

Changes for go-apps/meep-federation/go.mod: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -19,6 +19,7 @@ require (
		github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-service-mgmt-client v0.0.0
		github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-subscriptions v0.0.0
		github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-swagger-api-mgr v0.0.0
		github.com/google/uuid v1.2.0
		github.com/gorilla/handlers v1.5.1
		github.com/gorilla/mux v1.8.0
		github.com/prometheus/client_golang v1.9.0

go-apps/meep-federation/sbi/federation-sbi.go

+18 −6

Changes for go-apps/meep-federation/sbi/federation-sbi.go: 18 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -65,10 +65,14 @@ type FedSbi struct {
		mutex sync.Mutex
		}

		type SystemInfo struct {
		type SystemInfo struct { // MEC Federation implementation of sharing MEC Plateform details
		SystemId string
		SystemName string
		SystemProvider string
		TransportTypes string
		Protocol string
		SerializerType string
		Host string
		}

		var sbi *FedSbi
		@@ -197,10 +201,6 @@ func Stop() (err error) {

		// Delete FED Manager
		if sbi.federationMgr != nil {
		if sbi.messageBrokerStarted {
		sbi.federationMgr.StopFedMessageBrokerServer()
		sbi.messageBrokerStarted = false
		}
		err = sbi.federationMgr.DeleteFederationMgr()
		if err != nil {
		log.Error(err.Error())
		@@ -279,7 +279,7 @@ func RegisterSystemInfo(requestData SystemInfo) (responseData SystemInfo, err er
		}
		err = sbi.federationMgr.PublishMessageOnMessageBroker(s)
		if err != nil {
		log.Error("Failed to start message broker server: ", err)
		log.Error("Failed to publish message broker server: ", err)
		return responseData, err
		}

		@@ -288,6 +288,18 @@ func RegisterSystemInfo(requestData SystemInfo) (responseData SystemInfo, err er
		return responseData, nil
		}

		func DeregisterSystemInfo(systemId string) (err error) {
		log.Info(">>> DeregisterSystemInfo: ", systemId)

		err = sbi.federationMgr.DeleteFederationMgr()
		if err != nil {
		log.Error(err)
		return err
		}

		return nil
		}

		func convertSystemInfotoJson(systemInfo *SystemInfo) string {
		jsonInfo, err := json.Marshal(*systemInfo)
		if err != nil {

go-apps/meep-federation/server/federation.go

+204 −11

Changes for go-apps/meep-federation/server/federation.go: 204 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -38,7 +38,10 @@ import (
		scc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-sandbox-ctrl-client"
		smc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-service-mgmt-client"
		sm "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-subscriptions"
		//"github.com/gorilla/mux"

		uuid "github.com/google/uuid"

		"github.com/gorilla/mux"
		)

		const moduleName = "meep-federation"
		@@ -89,7 +92,9 @@ var sbxCtrlClient *scc.APIClient
		var registrationTicker *time.Ticker
		var subMgr *sm.SubscriptionMgr = nil

		var subscriptionExpiryMap = map[int][]int{}
		var subscriptionExpiryMap = map[int][]int{} // List of subscriptions
		var systemInfopMap = map[string]sbi.SystemInfo{} // List of discovered MEC Plateform
		var localSystemInfo sbi.SystemInfo // Local registration from MEO

		var mutex sync.Mutex
		var expiryTicker *time.Ticker
		@@ -163,6 +168,20 @@ func registerService(appInstanceId string) error {
		return nil
		}

		func getMecServices() ([]smc.ServiceInfo, error) {
		if svcMgmtClient == nil {
		err := errors.New("App Enablement Service Management REST API client not set")
		log.Error(err.Error())
		return nil, err
		}
		appServicesResponse, _, err := svcMgmtClient.MecServiceMgmtApi.ServicesGET(context.TODO(), nil)
		if err != nil {
		log.Error("Failed to retrieve mec services on platform ", err)
		return nil, err
		}
		return appServicesResponse, nil
		}

		func sendReadyConfirmation(appInstanceId string) error {
		var appReady asc.AppReadyConfirmation
		appReady.Indication = "READY"
		@@ -595,6 +614,7 @@ func errHandlerProblemDetails(w http.ResponseWriter, error string, code int) {
		fmt.Fprint(w, jsonResponse)
		}

		// FIXME To be rmoved if not used at the end
		func repopulateSubscriptionMap(key string, jsonInfo string, userData interface{}) error {
		log.Debug(">>> repopulateSubscriptionMap: key: ", key)
		log.Debug(">>> repopulateSubscriptionMap: jsonInfo: ", jsonInfo)
		@@ -630,14 +650,64 @@ func repopulateSubscriptionMap(key string, jsonInfo string, userData interface{}
		}

		func systeminfoByIdDELETE(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> systeminfoByIdDELETE: ", r)

		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
		w.WriteHeader(http.StatusOK)

		vars := mux.Vars(r)
		log.Debug("systeminfoByIdDELETE: vars: ", vars)

		systemIdParamStr := vars["systemId"]
		log.Debug("systeminfoByIdDELETE: systemIdParamStr: ", systemIdParamStr)

		if _, ok := systemInfopMap[systemIdParamStr]; !ok {
		err := errors.New("Unknown systemId: " + systemIdParamStr)
		errHandlerProblemDetails(w, err.Error(), http.StatusNotFound)
		return
		}

		err := sbi.DeregisterSystemInfo(systemIdParamStr)
		if err != nil {
		log.Error("Failed to de-register SystemInfo: ", err)
		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)
		return
		}
		log.Info("De-registration succeed")

		delete(systemInfopMap, systemIdParamStr)

		w.WriteHeader(http.StatusNoContent)
		}

		func systeminfoByIdGET(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> systeminfoByIdGET: ", r)

		w.Header().Set("Content-Type", "application/json; charset=UTF-8")

		vars := mux.Vars(r)
		log.Debug("systeminfoByIdGET: vars: ", vars)

		systemIdParamStr := vars["systemId"]
		log.Debug("systeminfoByIdGET: systemIdParamStr: ", systemIdParamStr)

		val, ok := systemInfopMap[systemIdParamStr]
		if len(systemInfopMap) == 0 \|\| !ok {
		w.WriteHeader(http.StatusNotFound)
		} else {
		var l = SystemInfo{
		SystemId: val.SystemId,
		SystemName: val.SystemName,
		SystemProvider: val.SystemProvider,
		}
		jsonResponse, err := json.Marshal(l)
		if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
		}
		fmt.Fprintf(w, string(jsonResponse))
		w.WriteHeader(http.StatusOK)
		}
		}

		func systeminfoByIdPATCH(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
		@@ -645,9 +715,58 @@ func systeminfoByIdPATCH(w http.ResponseWriter, r *http.Request) {
		}

		func systeminfoGET(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> systeminfoGET: ", r)

		w.Header().Set("Content-Type", "application/json; charset=UTF-8")

		// Validate query parameters
		u, _ := url.Parse(r.URL.String())
		q := u.Query()
		validParams := []string{"systemId", "systemName", "systemProvider"}
		err := validateQueryParams(q, validParams)
		if err != nil {
		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)
		return
		}

		systemId := q.Get("systemId")
		systemName := q.Get("systemName")
		systemProvider := q.Get("systemProvider")
		noFilter := systemId == "" && systemName == "" && systemProvider == ""
		log.Debug("systeminfoGET: noFilter: ", noFilter)

		if len(systemInfopMap) == 0 {
		w.WriteHeader(http.StatusNotFound)
		} else {
		l := []sbi.SystemInfo{}
		for _, val := range systemInfopMap {
		if noFilter {
		l = append(l, val)
		} else if systemId != "" && val.SystemId == systemId \|\| systemName != "" && val.SystemName == systemName \|\| systemProvider != "" && val.SystemProvider == systemProvider {
		l = append(l, val)
		} // else skip this item
		}
		if len(l) == 0 {
		w.WriteHeader(http.StatusNotFound)
		return
		}
		var systemInfos = []SystemInfo{}
		for _, val := range systemInfopMap {
		systemInfos = append(systemInfos, SystemInfo{
		SystemId: val.SystemId,
		SystemName: val.SystemName,
		SystemProvider: val.SystemProvider,
		})
		}
		jsonResponse, err := json.Marshal(systemInfos)
		if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
		}
		fmt.Fprint(w, string(jsonResponse))
		w.WriteHeader(http.StatusOK)
		}
		}

		func systeminfoPOST(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> systeminfoPOST: ", r)
		@@ -661,11 +780,12 @@ func systeminfoPOST(w http.ResponseWriter, r *http.Request) {
		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)
		return
		}
		log.Debug("systeminfoPOST: requestData: ", requestData)

		// Validating mandatory parameters in request
		if requestData.SystemId == "" {
		log.Error("Mandatory systemId parameter not present")
		errHandlerProblemDetails(w, "Mandatory attribute systemId is missing in the request body.", http.StatusBadRequest)
		if requestData.SystemId != "" {
		log.Error("Mandatory systemId parameter shall be absent")
		errHandlerProblemDetails(w, "Mandatory attribute systemId shall be absent in the request body.", http.StatusBadRequest)
		return
		}

		@@ -682,9 +802,13 @@ func systeminfoPOST(w http.ResponseWriter, r *http.Request) {
		}

		var s = sbi.SystemInfo{
		SystemId: requestData.SystemId,
		SystemId: uuid.New().String(),
		SystemName: requestData.SystemName,
		SystemProvider: requestData.SystemProvider,
		TransportTypes: "HTTP",
		Protocol: "REST_HTTP",
		SerializerType: "JSON",
		Host: hostUrl.String(),
		}
		responseData, err := sbi.RegisterSystemInfo(s)
		if err != nil {
		@@ -692,14 +816,21 @@ func systeminfoPOST(w http.ResponseWriter, r *http.Request) {
		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)
		return
		}
		log.Info("SBI Initialized")
		// Same data type
		localSystemInfo = responseData
		// Populate systemInfopMap
		systemInfopMap[responseData.SystemId] = responseData
		log.Debug("systeminfoPOST: new systemInfopMap: ", systemInfopMap)

		log.Info("Registration succeed: ", localSystemInfo)

		var response = SystemInfo{
		// Prepare & send response
		var l = SystemInfo{
		SystemId: responseData.SystemId,
		SystemName: responseData.SystemName,
		SystemProvider: responseData.SystemProvider,
		}
		jsonResponse := convertSystemInfotoJson(&response)
		jsonResponse := convertSystemInfotoJson(&l)
		w.WriteHeader(http.StatusOK)
		fmt.Fprint(w, jsonResponse)
		}
		@@ -735,12 +866,56 @@ func serviceGET(w http.ResponseWriter, r *http.Request) {
		}

		func servicesGET(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> servicesGET: ", r)

		w.Header().Set("Content-Type", "application/json; charset=UTF-8")

		vars := mux.Vars(r)
		log.Debug("servicesGET: vars: ", vars)

		systemIdParamStr := vars["systemId"]
		log.Debug("servicesGET: systemIdParamStr: ", systemIdParamStr)

		val, ok := systemInfopMap[systemIdParamStr]
		if len(systemInfopMap) == 0 \|\| !ok {
		w.WriteHeader(http.StatusNotFound)
		} else if localSystemInfo.SystemId == systemIdParamStr { // Local request
		log.Debug("servicesGET: Local request: ", systemIdParamStr)
		services, err := getMecServices()
		if err != nil {
		errHandlerProblemDetails(w, err.Error(), http.StatusInternalServerError)
		return
		}
		log.Debug("servicesGET: ", services)
		} else { // Forward to peer MEF
		// FIME FSCOM To do
		err := errors.New("servicesGET: Unknown system ID: " + val.SystemId)
		log.Debug(err.Error())
		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)
		return
		}

		w.WriteHeader(http.StatusOK)
		}

		func fedNotify(msg string) {
		fmt.Println(">>> fedNotify: ", msg)
		log.Debug(">>> fedNotify: ", msg)

		var systemInfo sbi.SystemInfo
		err := json.Unmarshal([]byte(msg), &systemInfo)
		if err != nil {
		log.Error(err.Error())
		return
		}

		if systemInfo.SystemId != localSystemInfo.SystemId { // Do not notify on our registration
		// Populate systemInfopMap
		systemInfopMap[systemInfo.SystemId] = systemInfo
		log.Debug("fedNotify: new systemInfopMap: ", systemInfopMap)

		// TODO Send notification if any subscription
		log.Debug("fedNotify: Send notification: ", systemInfo)
		}
		}

		/*
		@@ -824,3 +999,21 @@ func checkForExpiredSubscriptions() {
		// func computeElapseTime(expiryDeadline TimeStamp) *TimeStamp {
		// return &TimeStamp{NanoSeconds: 0, Seconds: int32(expiryDeadline.Seconds) - int32(time.Now().Unix())}
		// }

		func validateQueryParams(params url.Values, validParams []string) error {
		for param := range params {
		found := false
		for _, validParam := range validParams {
		if param == validParam {
		found = true
		break
		}
		}
		if !found {
		err := errors.New("Invalid query param: " + param)
		log.Error(err.Error())
		return err
		}
		}
		return nil
		}

go-apps/meep-federation/server/federation_test.go

+560 −1

File changed.

Preview size limit exceeded, changes collapsed.

go-packages/meep-federation-client/README.md

+1 −1

Changes for go-packages/meep-federation-client/README.md: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -18,7 +18,7 @@ import "./swagger"

		## Documentation for API Endpoints

		All URIs are relative to https://localhost/sandboxname/sandboxname/fed_enablement/v1
		All URIs are relative to https://localhost/sandboxname/fed_enablement/v1

		Class \| Method \| HTTP request \| Description
		------------ \| ------------- \| ------------- \| -------------