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)
130 runVesmgr(chVesagent, chSupervision, chXAppNotifications, chXAppSubscriptions)
133 func startVesagent(ch chan error) {
134 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"))
135 cmd.Stdout = os.Stdout
136 cmd.Stderr = os.Stderr
137 if err := cmd.Start(); err != nil {
138 logger.Error("vesmgr exiting, ves-agent start failed: %s", err)
143 logger.Info("ves-agent started with pid %d", cmd.Process.Pid)
144 vesagent.Pid = cmd.Process.Pid
145 vesagent.process = cmd.Process
147 // wait ves-agent exit and then post the error to the channel
154 func killVespa(process *os.Process) {
155 logger.Info("Killing vespa")
156 err := process.Kill()
158 logger.Error("Cannot kill vespa: %s", err.Error())
162 func queryXAppsStatus(appmgrUrl string, timeout time.Duration) ([]byte, error) {
164 logger.Info("query xAppStatus started, url %s", appmgrUrl)
165 req, err := http.NewRequest("GET", appmgrUrl, nil)
167 logger.Error("Failed to create a HTTP request: %s", err)
170 req.Header.Set("Content-Type", "application/json")
171 client := &http.Client{}
172 client.Timeout = time.Second * timeout
173 resp, err := client.Do(req)
175 logger.Error("Query xApp status failed: %s", err)
178 defer resp.Body.Close()
179 if resp.StatusCode == http.StatusOK {
180 body, err := ioutil.ReadAll(resp.Body)
182 logger.Error("Failed to read xApp status body: %s", err)
185 logger.Info("query xAppStatus completed")
188 logger.Error("Error from xApp status query: %s", resp.Status)
189 return nil, errors.New(resp.Status)
196 normalState state = iota
197 vespaTerminatingState state = iota
200 func queryConf() ([]byte, error) {
201 return queryXAppsStatus("http://"+appmgrDomain+":"+appmgrPort+appmgrXAppConfigPath,
205 func runVesmgr(chVesagent chan error, chSupervision chan chan string, chXAppNotifications chan []byte, chXAppSubscriptions chan subsChannel) {
207 logger.Info("vesmgr main loop ready")
208 mystate := normalState
209 var xappStatus []byte
213 case supervision := <-chSupervision:
214 logger.Info("vesmgr: supervision")
216 case xAppNotif := <-chXAppNotifications:
217 logger.Info("vesmgr: xApp notification")
218 logger.Info(string(xAppNotif))
220 * If xapp status query fails then we cannot create
221 * a new configuration and kill vespa.
222 * In that case we assume that
223 * the situation is fixed when the next
226 xappStatus, err = queryConf()
228 killVespa(vesagent.process)
229 mystate = vespaTerminatingState
231 case err := <-chVesagent:
233 case vespaTerminatingState:
234 logger.Info("Vesagent termination completed")
235 createConf(xappStatus)
236 startVesagent(chVesagent)
237 mystate = normalState
239 logger.Error("Vesagent exited: " + err.Error())
242 case isSubscribed := <-chXAppSubscriptions:
243 if isSubscribed.err != nil {
244 logger.Error("Failed to make xApp subscriptions, vesmgr exiting: %s", isSubscribed.err)
247 xappStatus, err = queryConf()
249 createConf(xappStatus)
250 startVesagent(chVesagent)