smoversion/stub/simulator.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2021: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package main
  22
  23 import (
  24         "encoding/csv"
  25         "encoding/json"
  26         "flag"
  27         "fmt"
  28         "math/rand"
  29         "net/http"
  30         "os"
  31         "strconv"
  32         "sync"
  33         "time"
  34
  35         "github.com/prometheus/client_golang/prometheus"
  36         "github.com/prometheus/client_golang/prometheus/promhttp"
  37
  38         "github.com/gorilla/mux"
  39         "oransc.org/usecase/oduclosedloop/messages"
  40
  41         log "github.com/sirupsen/logrus"
  42 )
  43
  44 const THRESHOLD_TPUT int = 7000
  45
  46 type SliceAssuranceInformation struct {
  47         duId                 string
  48         cellId               string
  49         sd                   int
  50         sst                  int
  51         metricName           string
  52         metricValue          int
  53         policyRatioId        string
  54         policyMaxRatio       int
  55         policyMinRatio       int
  56         policyDedicatedRatio int
  57 }
  58
  59 var (
  60         data           []*SliceAssuranceInformation
  61         messagesToSend []messages.Measurement
  62         metric         = prometheus.NewGaugeVec(prometheus.GaugeOpts{
  63                 Name: "rapp_slice_assurance_metric",
  64                 Help: "Duration of the last request call.",
  65         }, []string{"MeasurementTypeInstanceReference", "unit"})
  66 )
  67
  68 func loadData() {
  69         lines, err := GetCsvFromFile("test-data.csv")
  70         if err != nil {
  71                 panic(err)
  72         }
  73         for _, line := range lines {
  74                 sai := SliceAssuranceInformation{
  75                         duId:                 line[0],
  76                         cellId:               line[1],
  77                         sd:                   toInt(line[2]),
  78                         sst:                  toInt(line[3]),
  79                         metricName:           line[4],
  80                         metricValue:          toInt(line[5]),
  81                         policyRatioId:        line[6],
  82                         policyMaxRatio:       toInt(line[7]),
  83                         policyMinRatio:       toInt(line[8]),
  84                         policyDedicatedRatio: toInt(line[9]),
  85                 }
  86                 data = append(data, &sai)
  87                 metric.WithLabelValues(getMeasTypeInstanceRef(&sai), "kbit/s").Set(float64(THRESHOLD_TPUT))
  88         }
  89 }
  90
  91 func GetCsvFromFile(name string) ([][]string, error) {
  92         if csvFile, err := os.Open(name); err == nil {
  93                 defer csvFile.Close()
  94                 reader := csv.NewReader(csvFile)
  95                 reader.FieldsPerRecord = -1
  96                 if csvData, err := reader.ReadAll(); err == nil {
  97                         return csvData, nil
  98                 } else {
  99                         return nil, err
 100                 }
 101         } else {
 102                 return nil, err
 103         }
 104 }
 105
 106 func toInt(num string) int {
 107         res, err := strconv.Atoi(num)
 108         if err != nil {
 109                 return -1
 110         }
 111         return res
 112 }
 113
 114 func main() {
 115         rand.Seed(time.Now().UnixNano())
 116
 117         portSdnr := flag.Int("sdnr-port", 3904, "The port this SDNR stub will listen on")
 118         portDmaapMR := flag.Int("dmaap-port", 3905, "The port this Dmaap message router will listen on")
 119         flag.Parse()
 120
 121         loadData()
 122
 123         wg := new(sync.WaitGroup)
 124         wg.Add(2)
 125
 126         prometheus.Register(metric)
 127
 128         go func() {
 129
 130                 r := mux.NewRouter()
 131                 r.HandleFunc("/rests/data/network-topology:network-topology/topology=topology-netconf/node={NODE-ID}/yang-ext:mount/o-ran-sc-du-hello-world:network-function/distributed-unit-functions={O-DU-ID}", getSdnrResponseMessage).Methods(http.MethodGet)
 132                 r.HandleFunc("/rests/data/network-topology:network-topology/topology=topology-netconf/node={NODE-ID}/yang-ext:mount/o-ran-sc-du-hello-world:network-function/distributed-unit-functions={O-DU-ID}/radio-resource-management-policy-ratio={POLICY-ID}", updateRRMPolicyDedicatedRatio).Methods(http.MethodPut)
 133
 134                 fmt.Println("Starting SDNR stub on port: ", *portSdnr)
 135
 136                 log.Fatal(http.ListenAndServe(fmt.Sprintf(":%v", *portSdnr), r))
 137                 wg.Done()
 138         }()
 139
 140         go func() {
 141
 142                 r := mux.NewRouter()
 143                 r.Handle("/metrics", promhttp.Handler())
 144                 r.HandleFunc("/events/unauthenticated.VES_O_RAN_SC_HELLO_WORLD_PM_STREAMING_OUTPUT/myG/C1", sendDmaapMRMessages).Methods(http.MethodGet)
 145
 146                 fmt.Println("Starting DmaapMR stub on port: ", *portDmaapMR)
 147
 148                 log.Fatal(http.ListenAndServe(fmt.Sprintf(":%v", *portDmaapMR), r))
 149                 wg.Done()
 150         }()
 151
 152         wg.Wait()
 153 }
 154
 155 func getSdnrResponseMessage(w http.ResponseWriter, r *http.Request) {
 156         vars := mux.Vars(r)
 157         log.Info("Get messages for RRM Policy Ratio information for O-Du ID ", vars["O-DU-ID"])
 158
 159         distUnitFunctions := getDistributedUnitFunctionMessage(vars["O-DU-ID"])
 160
 161         respondWithJSON(w, http.StatusOK, distUnitFunctions)
 162 }
 163
 164 func getDistributedUnitFunctionMessage(oduId string) messages.ORanDuRestConf {
 165
 166         var policies []messages.RRMPolicyRatio
 167         keys := make(map[string]bool)
 168         for _, entry := range data {
 169                 if _, value := keys[entry.policyRatioId]; !value {
 170                         keys[entry.policyRatioId] = true
 171                         message := messages.RRMPolicyRatio{
 172
 173                                 Id:                      entry.policyRatioId,
 174                                 AdmState:                "locked",
 175                                 UserLabel:               entry.policyRatioId,
 176                                 RRMPolicyMaxRatio:       entry.policyMaxRatio,
 177                                 RRMPolicyMinRatio:       entry.policyMinRatio,
 178                                 RRMPolicyDedicatedRatio: entry.policyDedicatedRatio,
 179                                 ResourceType:            "prb",
 180                                 RRMPolicyMembers: []messages.RRMPolicyMember{
 181                                         {
 182                                                 MobileCountryCode:   "310",
 183                                                 MobileNetworkCode:   "150",
 184                                                 SliceDifferentiator: entry.sd,
 185                                                 SliceServiceType:    entry.sst,
 186                                         },
 187                                 },
 188                         }
 189                         policies = append(policies, message)
 190                 }
 191         }
 192
 193         var publicLandMobileNetworks []messages.PublicLandMobileNetworks
 194         for _, entry := range data {
 195                 publicLandMobileNetwork := messages.PublicLandMobileNetworks{
 196                         MobileCountryCode:   "310",
 197                         MobileNetworkCode:   "150",
 198                         SliceDifferentiator: entry.sd,
 199                         SliceServiceType:    entry.sst,
 200                 }
 201                 publicLandMobileNetworks = append(publicLandMobileNetworks, publicLandMobileNetwork)
 202         }
 203
 204         var supportedSnssaiSubcounterInstances []messages.SupportedSnssaiSubcounterInstances
 205         for _, entry := range data {
 206                 supportedSnssaiSubcounterInstance := messages.SupportedSnssaiSubcounterInstances{
 207                         SliceDifferentiator: entry.sd,
 208                         SliceServiceType:    entry.sst,
 209                 }
 210                 supportedSnssaiSubcounterInstances = append(supportedSnssaiSubcounterInstances, supportedSnssaiSubcounterInstance)
 211         }
 212
 213         cell := messages.Cell{
 214                 Id:             "cell-1",
 215                 LocalId:        1,
 216                 PhysicalCellId: 1,
 217                 BaseStationChannelBandwidth: messages.BaseStationChannelBandwidth{
 218                         Uplink:              83000,
 219                         Downlink:            80000,
 220                         SupplementaryUplink: 84000,
 221                 },
 222                 OperationalState:         "enabled",
 223                 TrackingAreaCode:         10,
 224                 AdmState:                 "unlocked",
 225                 PublicLandMobileNetworks: publicLandMobileNetworks,
 226                 SupportedMeasurements: []messages.SupportedMeasurements{
 227                         {
 228                                 PerformanceMeasurementType:         "o-ran-sc-du-hello-world:user-equipment-average-throughput-uplink",
 229                                 SupportedSnssaiSubcounterInstances: supportedSnssaiSubcounterInstances,
 230                         },
 231                         {
 232                                 PerformanceMeasurementType:         "o-ran-sc-du-hello-world:user-equipment-average-throughput-downlink",
 233                                 SupportedSnssaiSubcounterInstances: supportedSnssaiSubcounterInstances,
 234                         },
 235                 },
 236                 TrafficState: "active",
 237                 AbsoluteRadioFrequencyChannelNumber: messages.AbsoluteRadioFrequencyChannelNumber{
 238                         Uplink:              14000,
 239                         Downlink:            15000,
 240                         SupplementaryUplink: 14500,
 241                 },
 242                 UserLabel: "cell-1",
 243                 SynchronizationSignalBlock: messages.SynchronizationSignalBlock{
 244                         Duration:               2,
 245                         FrequencyChannelNumber: 12,
 246                         Periodicity:            10,
 247                         SubcarrierSpacing:      30,
 248                         Offset:                 3,
 249                 },
 250         }
 251
 252         distUnitFunction := messages.DistributedUnitFunction{
 253                 Id:               oduId,
 254                 OperationalState: "enabled",
 255                 AdmState:         "unlocked",
 256                 UserLabel:        oduId,
 257                 Cell: []messages.Cell{
 258                         cell,
 259                 },
 260                 RRMPolicyRatio: policies,
 261         }
 262
 263         duRRMPolicyRatio := messages.ORanDuRestConf{
 264                 DistributedUnitFunction: []messages.DistributedUnitFunction{
 265                         distUnitFunction,
 266                 },
 267         }
 268
 269         return duRRMPolicyRatio
 270 }
 271
 272 func updateRRMPolicyDedicatedRatio(w http.ResponseWriter, r *http.Request) {
 273         var policies struct {
 274                 RRMPolicies []messages.RRMPolicyRatio `json:"radio-resource-management-policy-ratio"`
 275         }
 276         decoder := json.NewDecoder(r.Body)
 277
 278         if err := decoder.Decode(&policies); err != nil {
 279                 respondWithError(w, http.StatusBadRequest, "Invalid request payload")
 280                 return
 281         }
 282         defer r.Body.Close()
 283
 284         prMessages := policies.RRMPolicies
 285         log.Infof("Post request to update RRMPolicyDedicatedRatio %+v", prMessages)
 286         findAndUpdatePolicy(prMessages)
 287         respondWithJSON(w, http.StatusOK, map[string]string{"status": "200"})
 288 }
 289
 290 func findAndUpdatePolicy(rRMPolicyRatio []messages.RRMPolicyRatio) {
 291         for _, policy := range rRMPolicyRatio {
 292                 for _, entry := range data {
 293                         if entry.policyRatioId == policy.Id {
 294                                 log.Infof("update Policy Dedicated Ratio: value for policy %+v\n Old value: %v New value: %v ", policy, entry.policyDedicatedRatio, policy.RRMPolicyDedicatedRatio)
 295                                 entry.policyDedicatedRatio = policy.RRMPolicyDedicatedRatio
 296                                 if entry.metricValue > THRESHOLD_TPUT {
 297                                         entry.metricValue = rand.Intn(THRESHOLD_TPUT)
 298                                 }
 299                                 messagesToSend = append(messagesToSend, generateMeasurementEntry(entry))
 300                         }
 301                 }
 302         }
 303 }
 304
 305 func respondWithError(w http.ResponseWriter, code int, message string) {
 306         respondWithJSON(w, code, map[string]string{"error": message})
 307 }
 308
 309 func respondWithJSON(w http.ResponseWriter, code int, payload interface{}) {
 310         response, _ := json.Marshal(payload)
 311
 312         w.Header().Set("Content-Type", "application/json")
 313         w.WriteHeader(code)
 314         w.Write(response)
 315 }
 316
 317 func sendDmaapMRMessages(w http.ResponseWriter, r *http.Request) {
 318         log.Info("Send Dmaap messages")
 319         entry := data[rand.Intn(5)]
 320
 321         changeMetric(entry)
 322
 323         maxTput := THRESHOLD_TPUT + 100
 324         randomTput := rand.Intn(maxTput-THRESHOLD_TPUT+1) + THRESHOLD_TPUT
 325         if randomTput%3 == 0 {
 326                 log.Info("Using tput value higher than THRESHOLD_TPUT ", randomTput)
 327                 entry.metricValue = randomTput
 328         }
 329         randomEventId := rand.Intn(10000)
 330         messagesToSend = append(messagesToSend, generateMeasurementEntry(entry))
 331
 332         message := messages.StdDefinedMessage{
 333                 Event: messages.Event{
 334                         CommonEventHeader: messages.CommonEventHeader{
 335                                 Domain:                  "stndDefined",
 336                                 EventId:                 "pm-1_16442" + strconv.Itoa(randomEventId),
 337                                 EventName:               "stndDefined_performanceMeasurementStreaming",
 338                                 EventType:               "performanceMeasurementStreaming",
 339                                 Sequence:                825,
 340                                 Priority:                "Low",
 341                                 ReportingEntityId:       "",
 342                                 ReportingEntityName:     "O-DU-1122",
 343                                 SourceId:                "",
 344                                 SourceName:              "O-DU-1122",
 345                                 StartEpochMicrosec:      1644252450000000,
 346                                 LastEpochMicrosec:       1644252480000000,
 347                                 NfNamingCode:            "SIM-O-DU",
 348                                 NfVendorName:            "O-RAN-SC SIM Project",
 349                                 StndDefinedNamespace:    "o-ran-sc-du-hello-world-pm-streaming-oas3",
 350                                 TimeZoneOffset:          "+00:00",
 351                                 Version:                 "4.1",
 352                                 VesEventListenerVersion: "7.2.1",
 353                         },
 354                         StndDefinedFields: messages.StndDefinedFields{
 355                                 StndDefinedFieldsVersion: "1.0",
 356                                 SchemaReference:          "https://gerrit.o-ran-sc.org/r/gitweb?p=scp/oam/modeling.git;a=blob_plain;f=data-model/oas3/experimental/o-ran-sc-du-hello-world-oas3.json;hb=refs/heads/master",
 357                                 Data: messages.Data{
 358                                         DataId:              "pm-1_1644252450",
 359                                         StartTime:           "2022-02-07T16:47:30.0Z",
 360                                         AdministrativeState: "unlocked",
 361                                         OperationalState:    "enabled",
 362                                         UserLabel:           "pm",
 363                                         JobTag:              "my-job-tag",
 364                                         GranularityPeriod:   30,
 365                                         Measurements:        messagesToSend,
 366                                 },
 367                         },
 368                 },
 369         }
 370
 371         fmt.Printf("Sending Dmaap message:\n %+v\n", message)
 372
 373         messageAsByteArray, _ := json.Marshal(message)
 374         response := [1]string{string(messageAsByteArray)}
 375
 376         time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
 377         respondWithJSON(w, http.StatusOK, response)
 378
 379         messagesToSend = nil
 380 }
 381
 382 func generateMeasurementEntry(entry *SliceAssuranceInformation) messages.Measurement {
 383
 384         measurementTypeInstanceReference := getMeasTypeInstanceRef(entry)
 385         meas := messages.Measurement{
 386
 387                 MeasurementTypeInstanceReference: measurementTypeInstanceReference,
 388                 Value:                            entry.metricValue,
 389                 Unit:                             "kbit/s",
 390         }
 391         return meas
 392 }
 393
 394 func getMeasTypeInstanceRef(entry *SliceAssuranceInformation) string {
 395         return "/o-ran-sc-du-hello-world:network-function/distributed-unit-functions[id='" + entry.duId + "']/cell[id='" + entry.cellId + "']/supported-measurements[performance-measurement-type='(urn:o-ran-sc:yang:o-ran-sc-du-hello-world?revision=2021-11-23)" + entry.metricName + "']/supported-snssai-subcounter-instances[slice-differentiator='" + strconv.Itoa(entry.sd) + "'][slice-service-type='" + strconv.Itoa(entry.sst) + "']"
 396 }
 397
 398 func changeMetric(entry *SliceAssuranceInformation) {
 399         min := -5
 400         max := 5
 401         tmpVal := rand.Intn(max-min) + min
 402         log.Infof("Adding %v to cell: %v, diff: %v ", tmpVal, entry.cellId, strconv.Itoa(entry.sd))
 403         metric.WithLabelValues(getMeasTypeInstanceRef(entry), "kbit/s").Add(float64(tmpVal))
 404
 405 }