2 * Copyright (c) 2019 AT&T Intellectual Property.
3 * Copyright (c) 2018-2019 Nokia.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
29 mdcloggo "gerrit.o-ran-sc.org/r/com/golog.git"
32 var appmgrDomain string
34 const appmgrXAppConfigPath = "/ric/v1/config"
35 const appmgrPort = "8080"
37 type VesAgent struct {
48 type subsChannel struct {
55 var logger *mdcloggo.MdcLogger
57 const vesmgrXappNotifPort = "8080"
58 const vesmgrXappNotifPath = "/vesmgr_xappnotif/"
59 const timeoutPostXAppSubscriptions = 5
62 logger, _ = mdcloggo.InitLogger("vesmgr")
65 func getMyIP() (myIP string, retErr error) {
66 addrs, err := net.InterfaceAddrs()
68 logger.Error("net.InterfaceAddrs failed: %s", err.Error())
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
83 func createConf(xappMetrics []byte) {
84 f, err := os.Create("/etc/ves-agent/ves-agent.yaml")
86 logger.Error("Cannot create vespa conf file: %s", err.Error())
91 createVespaConfig(f, xappMetrics)
92 logger.Info("Vespa config created")
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)
103 vesagent.name = "ves-agent"
104 logger.Info("vesmgrInit")
107 if vesmgr.myIPAddress, err = getMyIP(); err != nil || vesmgr.myIPAddress == "" {
108 logger.Error("Cannot get myIPAddress: IP %s, err %s", vesmgr.myIPAddress, err.Error())
113 appmgrDomain, ok = os.LookupEnv("VESMGR_APPMGRDOMAIN")
115 logger.Info("Using appmgrdomain %s", appmgrDomain)
117 appmgrDomain = "service-ricplt-appmgr-http.ricplt.svc.cluster.local"
118 logger.Info("Using default appmgrdomain %s", appmgrDomain)
120 chXAppSubscriptions := make(chan subsChannel)
121 chXAppNotifications := make(chan []byte)
122 chSupervision := make(chan chan string)
123 chVesagent := make(chan error)
125 listener, err := net.Listen("tcp", vesmgr.myIPAddress+":"+vesmgrXappNotifPort)
126 startHttpServer(listener, vesmgrXappNotifPath, chXAppNotifications, chSupervision)
128 subscribeXAppNotifications(chXAppSubscriptions)
131 startVesagent(chVesagent)
133 runVesmgr(chVesagent, chSupervision, chXAppNotifications, chXAppSubscriptions)
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)
146 logger.Info("ves-agent started with pid %d", cmd.Process.Pid)
147 vesagent.Pid = cmd.Process.Pid
148 vesagent.process = cmd.Process
150 // wait ves-agent exit and then post the error to the channel
157 func killVespa(process *os.Process) {
158 logger.Info("Killing vespa")
159 err := process.Kill()
161 logger.Error("Cannot kill vespa: %s", err.Error())
165 func queryXAppsStatus(appmgrUrl string, timeout time.Duration) ([]byte, error) {
167 logger.Info("query xAppStatus started, url %s", appmgrUrl)
168 req, err := http.NewRequest("GET", appmgrUrl, nil)
170 logger.Error("Failed to create a HTTP request: %s", err)
173 req.Header.Set("Content-Type", "application/json")
174 client := &http.Client{}
175 client.Timeout = time.Second * timeout
176 resp, err := client.Do(req)
178 logger.Error("Query xApp status failed: %s", err)
181 defer resp.Body.Close()
182 if resp.StatusCode == http.StatusOK {
183 body, err := ioutil.ReadAll(resp.Body)
185 logger.Error("Failed to read xApp status body: %s", err)
188 logger.Info("query xAppStatus completed")
191 logger.Error("Error from xApp status query: %s", resp.Status)
192 return nil, errors.New(resp.Status)
199 normalState state = iota
200 vespaTerminatingState state = iota
203 func queryConf() ([]byte, error) {
204 return queryXAppsStatus("http://"+appmgrDomain+":"+appmgrPort+appmgrXAppConfigPath,
208 func runVesmgr(chVesagent chan error, chSupervision chan chan string, chXAppNotifications chan []byte, chXAppSubscriptions chan subsChannel) {
210 logger.Info("vesmgr main loop ready")
211 mystate := normalState
212 var xappStatus []byte
215 case supervision := <-chSupervision:
216 logger.Info("vesmgr: supervision")
218 case xAppNotif := <-chXAppNotifications:
219 logger.Info("vesmgr: xApp notification")
220 logger.Info(string(xAppNotif))
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
229 xappStatus, err = queryConf()
231 killVespa(vesagent.process)
232 mystate = vespaTerminatingState
234 case err := <-chVesagent:
236 case vespaTerminatingState:
237 logger.Info("Vesagent termination completed")
238 createConf(xappStatus)
239 startVesagent(chVesagent)
240 mystate = normalState
242 logger.Error("Vesagent exited: " + err.Error())
245 case isSubscribed := <-chXAppSubscriptions:
246 if isSubscribed.err != nil {
247 logger.Error("Failed to make xApp subscriptions, vesmgr exiting: %s", isSubscribed.err)