Enhance GIS engine to support simu5g - Missing files (272f39e8) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-apps/meep-sandbox-ctrl/server/nem.go

0 → 100644

+149 −0

Changes for go-apps/meep-sandbox-ctrl/server/nem.go: 149 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.
		*/

		package server

		import (
		"context"
		"encoding/json"
		"errors"
		"fmt"
		"net/http"

		log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"
		nwcat "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-nw-simu-catalog"
		)

		func (s *SandboxCtrl) usesNem() bool {
		return s != nil && s.nwCatalog != nil && s.nwCatalog.Cfg().UsesNem()
		}

		func (s *SandboxCtrl) scenarioAlreadyActive() bool {
		if s == nil {
		return false
		}
		if s.nemInstance != nil {
		return true
		}
		return s.activeModel != nil && s.activeModel.Active
		}

		func (s *SandboxCtrl) createNemInstance(scenarioName string) error {
		if !s.usesNem() {
		return nil
		}
		ctx, cancel := context.WithTimeout(context.Background(), s.nwCatalog.Cfg().ActivateTimeout())
		defer cancel()

		req := nwcat.CreateInstanceRequest{
		SandboxID: s.sandboxName,
		ScenarioID: scenarioName,
		SimTimeLimit: nwcat.DefaultSimTimeLimit,
		}
		inst, err := s.nwCatalog.CreateInstance(ctx, req)
		if err != nil && errors.Is(err, nwcat.ErrNemConflict) {
		log.Warn("NEM instance already active for sandbox ", s.sandboxName, "; fetching status")
		inst, err = s.nwCatalog.GetInstance(ctx, s.sandboxName)
		}
		if err == nil && !nwcat.InstanceReady(inst) {
		state := ""
		if inst != nil {
		state = inst.State
		}
		err = fmt.Errorf("%w: HTTP 500: instance %s state %s", nwcat.ErrNemUnavailable, s.sandboxName, state)
		}
		if err != nil {
		s.deleteNemInstance()
		return err
		}

		s.mutex.Lock()
		s.nemInstance = inst
		s.mutex.Unlock()

		log.Info("NEM instance READY sandbox_id=", inst.SandboxID,
		" scenario_id=", inst.ScenarioID,
		" state=", inst.State,
		" slot=", inst.Slot,
		" bridge=", inst.Bridge.Target,
		" sim_host=", inst.Endpoints.SimHost,
		" traci_port=", inst.Endpoints.TraciPort,
		" rni_udp_port=", inst.Endpoints.RniUDPPort,
		" ue_telemetry_port=", inst.Endpoints.UeTelemetryPort)
		return nil
		}

		func (s *SandboxCtrl) deleteNemInstance() {
		if !s.usesNem() {
		return
		}
		ctx, cancel := context.WithTimeout(context.Background(), s.nwCatalog.Cfg().ActivateTimeout())
		defer cancel()

		err := s.nwCatalog.DeleteInstance(ctx, s.sandboxName, false)
		if err != nil {
		log.Error("Failed to delete NEM instance ", s.sandboxName, ": ", err.Error())
		} else {
		log.Info("NEM instance deleted sandbox_id=", s.sandboxName)
		}
		s.mutex.Lock()
		s.nemInstance = nil
		s.mutex.Unlock()
		}

		func (s *SandboxCtrl) writeNemActiveScenario(w http.ResponseWriter) bool {
		s.mutex.Lock()
		inst := s.nemInstance
		s.mutex.Unlock()
		if inst == nil {
		return false
		}
		name := inst.ScenarioID
		if name == "" {
		name = s.sandboxName
		}
		body, err := json.Marshal(struct {
		Name string `json:"name"`
		Deployment map[string]interface{} `json:"deployment"`
		}{
		Name: name,
		Deployment: map[string]interface{}{"domains": []interface{}{}},
		})
		if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return true
		}
		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
		w.WriteHeader(http.StatusOK)
		_, _ = fmt.Fprint(w, string(body))
		return true
		}

		func mapNemActivateError(w http.ResponseWriter, err error) {
		switch {
		case errors.Is(err, nwcat.ErrNemConflict):
		http.Error(w, nwcat.UserMessage(err), http.StatusConflict)
		case errors.Is(err, nwcat.ErrNemBadRequest):
		http.Error(w, nwcat.UserMessage(err), http.StatusBadRequest)
		default:
		log.Error("NEM instance create: ", err.Error())
		if nwcat.Unreachable(err) {
		http.Error(w, nwcat.ErrMsgNemNotRunning, http.StatusBadGateway)
		return
		}
		http.Error(w, nwcat.UserMessage(err), http.StatusInternalServerError)
		}
		}

go-apps/meep-simu5g-mobility-adapter/gis_url.go

0 → 100644

+124 −0

Changes for go-apps/meep-simu5g-mobility-adapter/gis_url.go: 124 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.
		*/

		package main

		import (
		"bytes"
		"context"
		"fmt"
		"net"
		"os"
		"os/exec"
		"strings"
		"time"

		log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"
		bridge "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-simu5g-bridge"
		)

		func resolveGISBaseURL(cfg *bridge.IntegrationConfig, flagURL, flagSandbox string) (string, error) {
		if u := strings.TrimRight(strings.TrimSpace(flagURL), "/"); u != "" {
		return u, nil
		}
		if u := cfg.GISBaseURL(); u != "" {
		return u, nil
		}
		if u := strings.TrimRight(strings.TrimSpace(os.Getenv("MEEP_GIS_BASE_URL")), "/"); u != "" {
		return u, nil
		}

		sandbox := strings.TrimSpace(flagSandbox)
		if sandbox == "" {
		sandbox = strings.TrimSpace(cfg.GIS.SandboxName)
		}
		if sandbox == "" {
		sandbox = strings.TrimSpace(os.Getenv("MEEP_SANDBOX_NAME"))
		}

		if sandbox != "" {
		if hostResolves("meep-gis-engine." + sandbox + ".svc.cluster.local") {
		return "http://meep-gis-engine." + sandbox + ".svc.cluster.local/gis/v1", nil
		}
		ip, err := kubectlGISClusterIP(sandbox)
		if err != nil {
		return "", fmt.Errorf("cannot reach meep-gis-engine in namespace %s (set gis.baseUrl to http://<clusterIP>/gis/v1): %v", sandbox, err)
		}
		log.Info("GIS ClusterIP for sandbox " + sandbox + " is " + ip)
		return "http://" + ip + "/gis/v1", nil
		}

		ns, ip, err := kubectlDiscoverSingleGIS()
		if err != nil {
		return "", fmt.Errorf("gis.baseUrl or gis.sandboxName required (or --gis-url / --sandbox): %v", err)
		}
		log.Info("discovered GIS Service in sandbox " + ns + " ClusterIP " + ip)
		return "http://" + ip + "/gis/v1", nil
		}

		func hostResolves(host string) bool {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		var r net.Resolver
		_, err := r.LookupHost(ctx, host)
		return err == nil
		}

		func kubectlGISClusterIP(namespace string) (string, error) {
		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
		defer cancel()
		cmd := exec.CommandContext(ctx, "kubectl", "get", "svc", "meep-gis-engine", "-n", namespace, "-o", "jsonpath={.spec.clusterIP}")
		out, err := cmd.Output()
		if err != nil {
		return "", err
		}
		ip := strings.TrimSpace(string(out))
		if net.ParseIP(ip) == nil {
		return "", fmt.Errorf("invalid ClusterIP %q", ip)
		}
		return ip, nil
		}

		func kubectlDiscoverSingleGIS() (namespace, clusterIP string, err error) {
		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
		defer cancel()
		cmd := exec.CommandContext(ctx, "kubectl", "get", "svc", "-A",
		"--field-selector", "metadata.name=meep-gis-engine",
		"-o", `jsonpath={range .items[*]}{.metadata.namespace}{" "}{.spec.clusterIP}{"\n"}{end}`)
		out, err := cmd.Output()
		if err != nil {
		return "", "", err
		}
		lines := bytes.Split(bytes.TrimSpace(out), []byte("\n"))
		var found [][2]string
		for _, line := range lines {
		f := strings.Fields(string(line))
		if len(f) != 2 {
		continue
		}
		if net.ParseIP(f[1]) == nil {
		continue
		}
		found = append(found, [2]string{f[0], f[1]})
		}
		if len(found) == 0 {
		return "", "", fmt.Errorf("no meep-gis-engine Service found")
		}
		if len(found) > 1 {
		return "", "", fmt.Errorf("multiple GIS engines; set gis.sandboxName")
		}
		return found[0][0], found[0][1], nil
		}

go-apps/meep-simu5g-mobility-adapter/gis_url_test.go

0 → 100644

+20 −0

Changes for go-apps/meep-simu5g-mobility-adapter/gis_url_test.go: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		package main

		import (
		"testing"

		bridge "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-simu5g-bridge"
		)

		func TestResolveGISBaseURLFlagOverridesConfig(t *testing.T) {
		cfg := &bridge.IntegrationConfig{
		GIS: bridge.GISOptions{BaseURL: "http://from-config/gis/v1"},
		}
		u, err := resolveGISBaseURL(cfg, "http://from-flag/gis/v1/", "")
		if err != nil {
		t.Fatal(err)
		}
		if u != "http://from-flag/gis/v1" {
		t.Fatalf("got %q", u)
		}
		}

go-apps/meep-simu5g-mobility-adapter/publisher/gis.go

0 → 100644

+198 −0

Changes for go-apps/meep-simu5g-mobility-adapter/publisher/gis.go: 198 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.
		*/

		// Package publisher posts Simu5G telemetry positions to meep-gis-engine.
		package publisher

		import (
		"context"
		"fmt"
		"net"
		"net/http"
		"strings"
		"sync"
		"time"

		gis "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-gis-engine-client"
		log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"
		bridge "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-simu5g-bridge"
		)

		const (
		gisHTTPTimeout = 5 * time.Second
		gisMaxAttempts = 3
		)

		// GISPublisher calls UpdateGeoDataByName for mapped UEs.
		type GISPublisher struct {
		client *gis.APIClient
		baseURL string
		firstOK sync.Map
		firstErr sync.Map
		lastPoa sync.Map
		outTotal uint64
		mapMisses uint64
		writeFails uint64
		}

		// NewGISPublisher builds a GIS REST client. baseURL is e.g. http://10.98.48.147/gis/v1.
		func NewGISPublisher(baseURL string) *GISPublisher {
		cfg := gis.NewConfiguration()
		cfg.BasePath = strings.TrimRight(baseURL, "/")
		cfg.HTTPClient = &http.Client{Timeout: gisHTTPTimeout}
		return &GISPublisher{
		client: gis.NewAPIClient(cfg),
		baseURL: cfg.BasePath,
		}
		}

		// BaseURL returns the GIS engine v1 prefix.
		func (p *GISPublisher) BaseURL() string {
		if p == nil {
		return ""
		}
		return p.baseURL
		}

		// GeoDataFromUE builds a Point (lon, lat) update with no path (SUMO owns motion).
		func GeoDataFromUE(assetName string, ue bridge.UELine) gis.GeoDataAsset {
		return gis.GeoDataAsset{
		AssetName: assetName,
		AssetType: "UE",
		SubType: "UE",
		Location: &gis.Point{
		Type_: "Point",
		Coordinates: []float32{float32(ue.LonDeg), float32(ue.LatDeg)},
		},
		Velocity: float32(ue.SpeedMps),
		}
		}

		// Probe warns when GIS has no active scenario or is unreachable.
		func (p *GISPublisher) Probe(ctx context.Context) {
		if p == nil {
		return
		}
		_, resp, err := p.client.GeospatialDataApi.GetAssetData(ctx, nil)
		detail := gisErrDetail(err, resp)
		if err == nil {
		log.Info("GIS reachable at " + p.baseURL)
		return
		}
		if strings.Contains(detail, "No active scenario") {
		log.Warn("GIS has no active scenario at " + p.baseURL + "; activate monaco-harbor in the sandbox UI or GIS writes will fail")
		return
		}
		log.Warn("GIS probe failed (" + p.baseURL + "): " + detail)
		}

		// PublishUE maps and POSTs one telemetry line. Unmapped IDs are logged once and dropped.
		func (p GISPublisher) PublishUE(ctx context.Context, cfg bridge.IntegrationConfig, ue bridge.UELine) {
		if p == nil \|\| cfg == nil {
		return
		}
		asset, ok := cfg.ResolveSandboxUeID(ue.ID, ue.IPv4)
		if !ok {
		p.mapMisses++
		if _, seen := p.firstErr.LoadOrStore("unmap:"+ue.ID, true); !seen {
		log.Warn("Unmapped telemetry ue=" + ue.ID + " ip=" + ue.IPv4 + "; drop (no GIS write)")
		}
		return
		}
		if err := p.updateWithRetry(ctx, asset, ue); err != nil {
		p.writeFails++
		key := "err:" + asset
		if _, seen := p.firstErr.LoadOrStore(key, true); !seen {
		log.Error("GIS update failed asset=" + asset + " ue=" + ue.ID + ": " + err.Error())
		}
		return
		}
		p.outTotal++
		if _, seen := p.firstOK.LoadOrStore(asset, true); !seen {
		log.Info("GIS published asset=" + asset + " lat=" + fmt.Sprintf("%g", ue.LatDeg) + " lon=" + fmt.Sprintf("%g", ue.LonDeg))
		}
		p.logPoaInRange(ctx, asset)
		}

		func (p *GISPublisher) logPoaInRange(ctx context.Context, asset string) {
		geo, resp, err := p.client.GeospatialDataApi.GetGeoDataByName(ctx, asset, nil)
		if err != nil {
		if _, seen := p.firstErr.LoadOrStore("get:"+asset, true); !seen {
		log.Warn("GIS GET poaInRange failed asset=" + asset + ": " + gisErrDetail(err, resp))
		}
		return
		}
		joined := strings.Join(geo.PoaInRange, ",")
		prev, ok := p.lastPoa.Load(asset)
		if ok && prev == joined {
		return
		}
		p.lastPoa.Store(asset, joined)
		log.Info("GIS poaInRange asset=" + asset + " poas=[" + joined + "]")
		}

		func (p *GISPublisher) updateWithRetry(ctx context.Context, assetName string, ue bridge.UELine) error {
		geo := GeoDataFromUE(assetName, ue)
		var last error
		for attempt := 0; attempt < gisMaxAttempts; attempt++ {
		if attempt > 0 {
		select {
		case <-ctx.Done():
		return ctx.Err()
		case <-time.After(time.Duration(100(1<<uint(attempt-1))) time.Millisecond):
		}
		}
		resp, err := p.client.GeospatialDataApi.UpdateGeoDataByName(ctx, assetName, geo)
		if err == nil {
		return nil
		}
		last = fmt.Errorf("%s", gisErrDetail(err, resp))
		if !retryableGIS(resp, err) {
		return last
		}
		}
		return last
		}

		func retryableGIS(resp *http.Response, err error) bool {
		if err != nil {
		if _, ok := err.(net.Error); ok {
		return true
		}
		}
		code := 0
		if resp != nil {
		code = resp.StatusCode
		} else if gerr, ok := err.(gis.GenericSwaggerError); ok {
		fmt.Sscanf(gerr.Error(), "%d", &code)
		}
		return code == 502 \|\| code == 503 \|\| code == 504
		}

		func gisErrDetail(err error, resp *http.Response) string {
		if err == nil {
		return ""
		}
		msg := err.Error()
		if gerr, ok := err.(gis.GenericSwaggerError); ok {
		if b := strings.TrimSpace(string(gerr.Body())); b != "" {
		msg = msg + ": " + b
		}
		} else if resp != nil && resp.Status != "" {
		msg = resp.Status + ": " + msg
		}
		return msg
		}

go-apps/meep-simu5g-mobility-adapter/publisher/gis_test.go

0 → 100644

+114 −0

Changes for go-apps/meep-simu5g-mobility-adapter/publisher/gis_test.go: 114 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		package publisher

		import (
		"context"
		"encoding/json"
		"io/ioutil"
		"net/http"
		"net/http/httptest"
		"sync/atomic"
		"testing"

		bridge "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-simu5g-bridge"
		)

		func TestGeoDataFromUEPointOrder(t *testing.T) {
		geo := GeoDataFromUE("ue-car-0", bridge.UELine{
		LatDeg: 43.728,
		LonDeg: 7.4216,
		SpeedMps: 13.8,
		ID: "MonacoHarbor.ue_car[0]",
		})
		if geo.AssetName != "ue-car-0" \|\| geo.AssetType != "UE" {
		t.Fatalf("asset %s %s", geo.AssetName, geo.AssetType)
		}
		if geo.Path != nil {
		t.Fatal("path must be omitted (SUMO owns motion)")
		}
		if geo.Location == nil \|\| geo.Location.Type_ != "Point" \|\| len(geo.Location.Coordinates) != 2 {
		t.Fatal("expected Point [lon,lat]")
		}
		if geo.Location.Coordinates[0] != float32(7.4216) \|\| geo.Location.Coordinates[1] != float32(43.728) {
		t.Fatalf("coordinates %v want lon,lat", geo.Location.Coordinates)
		}
		}

		func TestPublishUEPostsGeodata(t *testing.T) {
		var posts int32
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodGet {
		w.Header().Set("Content-Type", "application/json")
		if r.URL.Path == "/gis/v1/geodata/ue-car-0" {
		w.Write([]byte(`{"assetName":"ue-car-0","assetType":"UE","poaInRange":["poa-enb","poa-gnb"]}`))
		return
		}
		w.Write([]byte(`{"geoDataAssets":[]}`))
		return
		}
		if r.Method != http.MethodPost \|\| r.URL.Path != "/gis/v1/geodata/ue-car-0" {
		t.Errorf("unexpected %s %s", r.Method, r.URL.Path)
		w.WriteHeader(http.StatusNotFound)
		return
		}
		body, _ := ioutil.ReadAll(r.Body)
		var geo map[string]interface{}
		if err := json.Unmarshal(body, &geo); err != nil {
		t.Errorf("json: %v", err)
		}
		if geo["assetName"] != "ue-car-0" {
		t.Errorf("assetName %v", geo["assetName"])
		}
		atomic.AddInt32(&posts, 1)
		w.WriteHeader(http.StatusOK)
		}))
		defer srv.Close()

		pub := NewGISPublisher(srv.URL + "/gis/v1")
		cfg := &bridge.IntegrationConfig{
		UEMappings: []bridge.UEMapping{{
		UeID: "ue-car-0",
		SandboxIPCidr: "10.0.0.1/32",
		Simu5gModulePath: "MonacoHarbor.ue_car[0]",
		}},
		}
		pub.PublishUE(context.Background(), cfg, bridge.UELine{
		ID: "MonacoHarbor.ue_car[0]",
		IPv4: "10.0.0.1",
		LatDeg: 43.73,
		LonDeg: 7.42,
		})
		if atomic.LoadInt32(&posts) != 1 {
		t.Fatalf("posts %d", posts)
		}

		pub.PublishUE(context.Background(), cfg, bridge.UELine{ID: "MonacoHarbor.unknown", IPv4: "9.9.9.9"})
		if atomic.LoadInt32(&posts) != 1 {
		t.Fatal("unmapped UE must not POST")
		}
		}

		func TestUpdateRetryOn503(t *testing.T) {
		var hits int32
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte(`{"geoDataAssets":[]}`))
		return
		}
		n := atomic.AddInt32(&hits, 1)
		if n == 1 {
		w.WriteHeader(http.StatusServiceUnavailable)
		return
		}
		w.WriteHeader(http.StatusOK)
		}))
		defer srv.Close()
		pub := NewGISPublisher(srv.URL + "/gis/v1")
		err := pub.updateWithRetry(context.Background(), "ue-car-0", bridge.UELine{LatDeg: 43, LonDeg: 7})
		if err != nil {
		t.Fatal(err)
		}
		if atomic.LoadInt32(&hits) != 2 {
		t.Fatalf("hits %d", hits)
		}
		}