cmd/vesmgr/vesmgr.go

   1 /*
   2  *  Copyright (c) 2019 AT&T Intellectual Property.
   3  *  Copyright (c) 2018-2019 Nokia.
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  */
  17
  18 package main
  19
  20 import (
  21         "errors"
  22         "io/ioutil"
  23         "net"
  24         "net/http"
  25         "os"
  26         "os/exec"
  27         "time"
  28
  29         mdcloggo "gerrit.o-ran-sc.org/r/com/golog.git"
  30 )
  31
  32 var appmgrDomain string
  33
  34 const appmgrXAppConfigPath = "/ric/v1/config"
  35 const appmgrPort = "8080"
  36
  37 type VesAgent struct {
  38         Pid     int
  39         name    string
  40         process *os.Process
  41 }
  42
  43 type VesMgr struct {
  44         myIPAddress  string
  45         appmgrSubsId string
  46 }
  47
  48 type subsChannel struct {
  49         subscribed bool
  50         err        error
  51 }
  52
  53 var vesagent VesAgent
  54 var vesmgr VesMgr
  55 var logger *mdcloggo.MdcLogger
  56
  57 const vesmgrXappNotifPort = "8080"
  58 const vesmgrXappNotifPath = "/vesmgr_xappnotif/"
  59 const timeoutPostXAppSubscriptions = 5
  60
  61 func init() {
  62         logger, _ = mdcloggo.InitLogger("vesmgr")
  63 }
  64
  65 func getMyIP() (myIP string, retErr error) {
  66         addrs, err := net.InterfaceAddrs()
  67         if err != nil {
  68                 logger.Error("net.InterfaceAddrs failed: %s", err.Error())
  69                 return "", err
  70         }
  71         for _, addr := range addrs {
  72                 // check the address type and if it is not a loopback take it
  73                 if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
  74                         if ipnet.IP.To4() != nil {
  75                                 logger.Info("My IP Address: %s", ipnet.IP.String())
  76                                 return ipnet.IP.String(), nil
  77                         }
  78                 }
  79         }
  80         return "", nil
  81 }
  82
  83 func createConf(xappMetrics []byte) {
  84         f, err := os.Create("/etc/ves-agent/ves-agent.yaml")
  85         if err != nil {
  86                 logger.Error("Cannot create vespa conf file: %s", err.Error())
  87                 os.Exit(1)
  88         }
  89         defer f.Close()
  90
  91         createVespaConfig(f, xappMetrics)
  92         logger.Info("Vespa config created")
  93 }
  94
  95 func subscribeXAppNotifications(chSubscriptions chan subsChannel) {
  96         xappNotifUrl := "http://" + vesmgr.myIPAddress + ":" + vesmgrXappNotifPort + vesmgrXappNotifPath
  97         subsUrl := "http://" + appmgrDomain + ":" + appmgrPort + appmgrSubsPath
  98         go subscribexAppNotifications(xappNotifUrl, chSubscriptions, timeoutPostXAppSubscriptions, subsUrl)
  99         logger.Info("xApp notifications subscribed from %s", subsUrl)
 100 }
 101
 102 func vesmgrInit() {
 103         vesagent.name = "ves-agent"
 104         logger.Info("vesmgrInit")
 105
 106         var err error
 107         if vesmgr.myIPAddress, err = getMyIP(); err != nil || vesmgr.myIPAddress == "" {
 108                 logger.Error("Cannot get myIPAddress: IP %s, err %s", vesmgr.myIPAddress, err.Error())
 109                 return
 110         }
 111
 112         var ok bool
 113         appmgrDomain, ok = os.LookupEnv("VESMGR_APPMGRDOMAIN")
 114         if ok {
 115                 logger.Info("Using appmgrdomain %s", appmgrDomain)
 116         } else {
 117                 appmgrDomain = "service-ricplt-appmgr-http.ricplt.svc.cluster.local"
 118                 logger.Info("Using default appmgrdomain %s", appmgrDomain)
 119         }
 120         chXAppSubscriptions := make(chan subsChannel)
 121         chXAppNotifications := make(chan []byte)
 122         chSupervision := make(chan chan string)
 123         chVesagent := make(chan error)
 124
 125         listener, err := net.Listen("tcp", vesmgr.myIPAddress+":"+vesmgrXappNotifPort)
 126         startHttpServer(listener, vesmgrXappNotifPath, chXAppNotifications, chSupervision)
 127
 128         subscribeXAppNotifications(chXAppSubscriptions)
 129
 130         createConf([]byte{})
 131         startVesagent(chVesagent)
 132
 133         runVesmgr(chVesagent, chSupervision, chXAppNotifications, chXAppSubscriptions)
 134 }
 135
 136 func startVesagent(ch chan error) {
 137         cmd := exec.Command(vesagent.name, "-i", os.Getenv("VESMGR_HB_INTERVAL"), "-m", os.Getenv("VESMGR_MEAS_INTERVAL"), "--Measurement.Prometheus.Address", os.Getenv("VESMGR_PROMETHEUS_ADDR"))
 138         cmd.Stdout = os.Stdout
 139         cmd.Stderr = os.Stderr
 140         if err := cmd.Start(); err != nil {
 141                 logger.Error("vesmgr exiting, ves-agent start failed: %s", err)
 142                 go func() {
 143                         ch <- err
 144                 }()
 145         } else {
 146                 logger.Info("ves-agent started with pid %d", cmd.Process.Pid)
 147                 vesagent.Pid = cmd.Process.Pid
 148                 vesagent.process = cmd.Process
 149                 go func() {
 150                         // wait ves-agent exit and then post the error to the channel
 151                         err := cmd.Wait()
 152                         ch <- err
 153                 }()
 154         }
 155 }
 156
 157 func killVespa(process *os.Process) {
 158         logger.Info("Killing vespa")
 159         err := process.Kill()
 160         if err != nil {
 161                 logger.Error("Cannot kill vespa: %s", err.Error())
 162         }
 163 }
 164
 165 func queryXAppsStatus(appmgrUrl string, timeout time.Duration) ([]byte, error) {
 166
 167         logger.Info("query xAppStatus started, url %s", appmgrUrl)
 168         req, err := http.NewRequest("GET", appmgrUrl, nil)
 169         if err != nil {
 170                 logger.Error("Failed to create a HTTP request: %s", err)
 171                 return nil, err
 172         }
 173         req.Header.Set("Content-Type", "application/json")
 174         client := &http.Client{}
 175         client.Timeout = time.Second * timeout
 176         resp, err := client.Do(req)
 177         if err != nil {
 178                 logger.Error("Query xApp status failed: %s", err)
 179                 return nil, err
 180         }
 181         defer resp.Body.Close()
 182         if resp.StatusCode == http.StatusOK {
 183                 body, err := ioutil.ReadAll(resp.Body)
 184                 if err != nil {
 185                         logger.Error("Failed to read xApp status body: %s", err)
 186                         return nil, err
 187                 }
 188                 logger.Info("query xAppStatus completed")
 189                 return body, nil
 190         } else {
 191                 logger.Error("Error from xApp status query: %s", resp.Status)
 192                 return nil, errors.New(resp.Status)
 193         }
 194 }
 195
 196 type state int
 197
 198 const (
 199         normalState           state = iota
 200         vespaTerminatingState state = iota
 201 )
 202
 203 func queryConf() ([]byte, error) {
 204         return queryXAppsStatus("http://"+appmgrDomain+":"+appmgrPort+appmgrXAppConfigPath,
 205                 10*time.Second)
 206 }
 207
 208 func runVesmgr(chVesagent chan error, chSupervision chan chan string, chXAppNotifications chan []byte, chXAppSubscriptions chan subsChannel) {
 209
 210         logger.Info("vesmgr main loop ready")
 211         mystate := normalState
 212         var xappStatus []byte
 213         for {
 214                 select {
 215                 case supervision := <-chSupervision:
 216                         logger.Info("vesmgr: supervision")
 217                         supervision <- "OK"
 218                 case xAppNotif := <-chXAppNotifications:
 219                         logger.Info("vesmgr: xApp notification")
 220                         logger.Info(string(xAppNotif))
 221                         /*
 222                          * If xapp status query fails then we cannot create
 223                          * a new configuration and kill vespa.
 224                          * In that case we assume that
 225                          * the situation is fixed when the next
 226                          * xapp notif comes
 227                          */
 228                         var err error
 229                         xappStatus, err = queryConf()
 230                         if err == nil {
 231                                 killVespa(vesagent.process)
 232                                 mystate = vespaTerminatingState
 233                         }
 234                 case err := <-chVesagent:
 235                         switch mystate {
 236                         case vespaTerminatingState:
 237                                 logger.Info("Vesagent termination completed")
 238                                 createConf(xappStatus)
 239                                 startVesagent(chVesagent)
 240                                 mystate = normalState
 241                         default:
 242                                 logger.Error("Vesagent exited: " + err.Error())
 243                                 os.Exit(1)
 244                         }
 245                 case isSubscribed := <-chXAppSubscriptions:
 246                         if isSubscribed.err != nil {
 247                                 logger.Error("Failed to make xApp subscriptions, vesmgr exiting: %s", isSubscribed.err)
 248                                 os.Exit(1)
 249                         }
 250                 }
 251         }
 252 }