lib/agent/xds-low-collector.go

   1 /*
   2  * Copyright (C) 2019 "IoT.bzh"
   3  * Author Sebastien Douheret <sebastien@iot.bzh>
   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 agent
  19
  20 import (
  21         "fmt"
  22         "io"
  23         "strings"
  24         "time"
  25
  26         "gerrit.automotivelinux.org/gerrit/src/xds/xds-agent.git/lib/aglafb"
  27         common "gerrit.automotivelinux.org/gerrit/src/xds/xds-common.git"
  28         uuid "github.com/satori/go.uuid"
  29 )
  30
  31 // XdsLowCollector .
  32 type XdsLowCollector struct {
  33         *Context
  34         ID                 string
  35         BaseURL            string
  36         ConnRetry          int
  37         Connected          bool
  38         Disabled           bool
  39         DefaultPlugins     []string
  40         DefaultCollectTime int
  41         // Private fields
  42         client      *common.HTTPClient
  43         logOut      io.Writer
  44         cbOnConnect OnConnectedXdsAlcCB
  45 }
  46
  47 // OnConnectedXdsAlcCB connect callback
  48 type OnConnectedXdsAlcCB func(svr *XdsLowCollector) error
  49
  50 // NewXdsLowCollector creates an instance of XdsLowCollector
  51 func NewXdsLowCollector(ctx *Context) *XdsLowCollector {
  52         return &XdsLowCollector{
  53                 Context:   ctx,
  54                 ID:        "XdsAlc-" + uuid.NewV1().String(),
  55                 BaseURL:   ctx.Config.FileConf.ProfileConf.XDSLowCollector.URL,
  56                 ConnRetry: ctx.Config.FileConf.ProfileConf.XDSLowCollector.ConnRetry,
  57                 Connected: false,
  58                 Disabled:  false,
  59                 logOut:    ctx.Log.Out,
  60                 DefaultPlugins: []string{
  61                         "cpu",
  62                         "memory",
  63                         // SEB "processes",
  64                         //"cpufreq",
  65                         //"thermal",
  66                         //"systemd_journal",
  67                         // SEB "systemd_file",
  68                 },
  69                 DefaultCollectTime: 5,
  70         }
  71 }
  72
  73 // Connect Establish HTTP connection with XDS Low Collector Dameon
  74 func (xs *XdsLowCollector) Connect() error {
  75         var err error
  76         var retry int
  77
  78         xs.Disabled = false
  79         xs.Connected = false
  80
  81         err = nil
  82         for retry = xs.ConnRetry; retry > 0; retry-- {
  83                 if err = xs._CreateConnectHTTP(); err == nil {
  84                         break
  85                 }
  86                 if retry == xs.ConnRetry {
  87                         // Notify only on the first conn error
  88                         // doing that avoid 2 notifs (conn false; conn true) on startup
  89                         xs._NotifyState()
  90                 }
  91                 xs.Log.Infof("Establishing connection to XDS Low Collector daemon (retry %d/%d)", retry, xs.ConnRetry)
  92                 time.Sleep(time.Second)
  93         }
  94         if retry == 0 {
  95                 // FIXME: re-use _Reconnect to wait longer in background
  96                 return fmt.Errorf("Connection to XDS Low Collector daemon failure")
  97         }
  98         if err != nil {
  99                 return err
 100         }
 101
 102         // Check HTTP connection and establish WS connection
 103         err = xs._Connect(false)
 104
 105         return err
 106 }
 107
 108 // ConnectOn Register a callback on events reception
 109 func (xs *XdsLowCollector) ConnectOn(f OnConnectedXdsAlcCB) error {
 110         xs.cbOnConnect = f
 111         return nil
 112 }
 113
 114 // GetVersion Send Get request to retrieve XDS Low Collector version
 115 func (xs *XdsLowCollector) GetVersion(res interface{}) error {
 116         // FIXME add suffix URLSuffix in common HTTP client lib instead of _BuildURL
 117         return xs.client.Get(xs._BuildURL("/version"), &res)
 118 }
 119
 120 // Init Initialize collector plugins
 121 func (xs *XdsLowCollector) Init() error {
 122         var err error
 123
 124         // Directly send config in order to init and config plugins
 125
 126         type alcCfgPluginT struct {
 127                 Plugin string `json:"plugin"`
 128                 Config string `json:"config"`
 129         }
 130
 131         cfg := []alcCfgPluginT{}
 132
 133         for _, p := range xs.DefaultPlugins {
 134                 cfg = append(cfg, alcCfgPluginT{
 135                         Plugin: p,
 136                         Config: "default",
 137                 })
 138         }
 139
 140         res := aglafb.NewAfbReply()
 141         xs.Log.Debugf("Low Collector /config %v", cfg)
 142         err = xs.client.Post(xs._BuildURL("/config"), cfg, res)
 143
 144         if err == nil && !res.Success() {
 145                 err = res.GetError()
 146         }
 147
 148         return err
 149 }
 150
 151 // Start data collection
 152 func (xs *XdsLowCollector) Start(time int) error {
 153         var err error
 154
 155         // TODO - SEB : support start one or multiple plugins
 156
 157         if time == 0 {
 158                 time = xs.DefaultCollectTime
 159         }
 160
 161         type alcStartT struct {
 162                 Plugin string `json:"plugin"`
 163                 Time   int    `json:"time"`
 164         }
 165
 166         // TODO SEB : allow to start only 1 plugin
 167         allInOne := true
 168         if allInOne {
 169
 170                 cfg := []alcStartT{}
 171                 for _, p := range xs.DefaultPlugins {
 172                         cfg = append(cfg, alcStartT{Plugin: p, Time: time})
 173                 }
 174
 175                 res := aglafb.NewAfbReply()
 176                 xs.Log.Debugf("Low Collector /start %v", cfg)
 177                 err = xs.client.Post(xs._BuildURL("/start"), cfg, res)
 178
 179                 if err == nil && !res.Success() {
 180                         err = res.GetError()
 181                 }
 182         } else {
 183                 for _, p := range xs.DefaultPlugins {
 184                         cfg := alcStartT{Plugin: p, Time: time}
 185
 186                         res := aglafb.NewAfbReply()
 187                         xs.Log.Debugf("Low Collector /start %v", cfg)
 188                         err = xs.client.Post(xs._BuildURL("/start"), cfg, res)
 189                         if err != nil {
 190                                 return err
 191                         }
 192                         if !res.Success() {
 193                                 return res.GetError()
 194                         }
 195                 }
 196         }
 197
 198         return err
 199 }
 200
 201 // Stop data collection
 202 func (xs *XdsLowCollector) Stop() error {
 203
 204         // TODO - SEB : support start one or multiple plugins
 205
 206         type alcStopT struct {
 207                 Plugin []string `json:"plugin"`
 208         }
 209
 210         cfg := alcStopT{}
 211         for _, p := range xs.DefaultPlugins {
 212                 cfg.Plugin = append(cfg.Plugin, p)
 213         }
 214
 215         res := aglafb.NewAfbReply()
 216         xs.Log.Debugf("Low Collector /stop %v", cfg)
 217         err := xs.client.Post(xs._BuildURL("/stop"), cfg, res)
 218
 219         if err == nil && !res.Success() {
 220                 err = res.GetError()
 221         }
 222
 223         return err
 224 }
 225
 226 // Read a single data of a specific plugin
 227 func (xs *XdsLowCollector) Read(data interface{}) error {
 228         return fmt.Errorf("No implemented")
 229 }
 230
 231 /***
 232 ** Private functions
 233 ***/
 234
 235 // _BuildURL .
 236 func (xs *XdsLowCollector) _BuildURL(url string) string {
 237         return url + "?token=HELLO&uuid=magic"
 238 }
 239
 240 // Create HTTP client
 241 func (xs *XdsLowCollector) _CreateConnectHTTP() error {
 242         var err error
 243         // FIXME SEB - Client key not in header but in cookie
 244         // temporary workaround: used _BuildURL to append uuid=magic in URL
 245         // map[Set-Cookie:[x-afb-uuid-5678=2b185cc3-276b-4097-91fa-d607eaf937e6; Path=/api; Max-Age=32000000; ...
 246         //port := strings.Split(xs.BaseURL, ":")[2]
 247         //"x-afb-uuid-" + port
 248
 249         xs.client, err = common.HTTPNewClient(xs.BaseURL,
 250                 common.HTTPClientConfig{
 251                         //HeaderClientKeyName: "Xds-Sid",
 252                         HeaderAPIKeyName: "token",
 253                         Apikey:           "HELLO",
 254                         URLPrefix:        "/api/alc",
 255                         CsrfDisable:      true,
 256                         LogOut:           xs.logOut,
 257                         LogPrefix:        "XDSALC: ",
 258                         LogLevel:         common.HTTPLogLevelWarning,
 259                 })
 260
 261         xs.client.SetLogLevel(xs.Log.Level.String())
 262
 263         if err != nil {
 264                 msg := ": " + err.Error()
 265                 if strings.Contains(err.Error(), "connection refused") {
 266                         msg = fmt.Sprintf("(url: %s)", xs.BaseURL)
 267                 }
 268                 return fmt.Errorf("ERROR: cannot connect to XDS Low Collector %s", msg)
 269         }
 270         if xs.client == nil {
 271                 return fmt.Errorf("ERROR: cannot connect to XDS Low Collector (null client)")
 272         }
 273
 274         return nil
 275 }
 276
 277 // _Connect Established HTTP and WS connection
 278 func (xs *XdsLowCollector) _Connect(reConn bool) error {
 279         var res interface{}
 280         if err := xs.client.Get(xs._BuildURL("/ping"), &res); err != nil {
 281
 282                 // SEB FIXME tempo Hack
 283                 time.Sleep(time.Microsecond * 300)
 284                 if err := xs.client.Get(xs._BuildURL("/ping"), &res); err != nil {
 285                         // SEB Hack tempo
 286                         // xs.Connected = false
 287                         // if !reConn {
 288                         //      xs._NotifyState()
 289                         // }
 290                         // return err
 291                 }
 292         }
 293
 294         xs.Connected = true
 295
 296         // Call OnConnect callback
 297         if xs.cbOnConnect != nil {
 298                 xs.cbOnConnect(xs)
 299         }
 300
 301         xs._NotifyState()
 302         return nil
 303 }
 304
 305 // _NotifyState Send event to notify changes
 306 func (xs *XdsLowCollector) _NotifyState() {
 307
 308         /* TODO
 309         evSts := xaapiv1.ServerCfg{
 310                 ID:         xs.ID,
 311                 URL:        xs.BaseURL,
 312                 APIURL:     xs.APIURL,
 313                 PartialURL: xs.PartialURL,
 314                 ConnRetry:  xs.ConnRetry,
 315                 Connected:  xs.Connected,
 316         }
 317         if err := xs.events.Emit(xaapiv1.EVTServerConfig, evSts, ""); err != nil {
 318                 xs.Log.Warningf("Cannot notify XdsServer state change: %v", err)
 319         }
 320         */
 321 }