X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=lib%2Fagent%2Fxdsserver.go;h=32656cf429e0b8b952f22bfd2e972bf0346abb33;hb=89563054ea6305c7270dcc6ab6fa5b286eb5f742;hp=e2c38c1bbecbf1780cdbb0dec5984a2160464ba5;hpb=4e9af3723740f16f3843a68508b6e933ea871b98;p=src%2Fxds%2Fxds-agent.git diff --git a/lib/agent/xdsserver.go b/lib/agent/xdsserver.go index e2c38c1..32656cf 100644 --- a/lib/agent/xdsserver.go +++ b/lib/agent/xdsserver.go @@ -1,3 +1,20 @@ +/* + * Copyright (C) 2017 "IoT.bzh" + * Author Sebastien Douheret + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package agent import ( @@ -11,9 +28,10 @@ import ( "time" "github.com/gin-gonic/gin" - "github.com/iotbzh/xds-agent/lib/apiv1" + "github.com/iotbzh/xds-agent/lib/xaapiv1" "github.com/iotbzh/xds-agent/lib/xdsconfig" common "github.com/iotbzh/xds-common/golib" + "github.com/iotbzh/xds-server/lib/xsapiv1" uuid "github.com/satori/go.uuid" sio_client "github.com/sebd71/go-socket.io-client" ) @@ -28,7 +46,7 @@ type XdsServer struct { ConnRetry int Connected bool Disabled bool - ServerConfig *XdsServerConfig + ServerConfig *xsapiv1.APIConfig // Events management CBOnError func(error) @@ -41,80 +59,7 @@ type XdsServer struct { ioSock *sio_client.Client logOut io.Writer apiRouter *gin.RouterGroup -} - -// XdsServerConfig Data return by GET /config -type XdsServerConfig struct { - ID string `json:"id"` - Version string `json:"version"` - APIVersion string `json:"apiVersion"` - VersionGitTag string `json:"gitTag"` - SupportedSharing map[string]bool `json:"supportedSharing"` - Builder XdsBuilderConfig `json:"builder"` -} - -// XdsBuilderConfig represents the builder container configuration -type XdsBuilderConfig struct { - IP string `json:"ip"` - Port string `json:"port"` - SyncThingID string `json:"syncThingID"` -} - -// XdsFolderType XdsServer folder type -type XdsFolderType string - -const ( - // XdsTypePathMap Path Mapping folder type - XdsTypePathMap = "PathMap" - // XdsTypeCloudSync Cloud synchronization (AKA syncthing) folder type - XdsTypeCloudSync = "CloudSync" - // XdsTypeCifsSmb CIFS (AKA samba) folder type - XdsTypeCifsSmb = "CIFS" -) - -// XdsFolderConfig XdsServer folder config -type XdsFolderConfig struct { - ID string `json:"id"` - Label string `json:"label"` - ClientPath string `json:"path"` - Type XdsFolderType `json:"type"` - Status string `json:"status"` - IsInSync bool `json:"isInSync"` - DefaultSdk string `json:"defaultSdk"` - ClientData string `json:"clientData"` // free form field that can used by client - - // Specific data depending on which Type is used - DataPathMap XdsPathMapConfig `json:"dataPathMap,omitempty"` - DataCloudSync XdsCloudSyncConfig `json:"dataCloudSync,omitempty"` -} - -// XdsPathMapConfig Path mapping specific data -type XdsPathMapConfig struct { - ServerPath string `json:"serverPath"` - CheckFile string `json:"checkFile"` - CheckContent string `json:"checkContent"` -} - -// XdsCloudSyncConfig CloudSync (AKA Syncthing) specific data -type XdsCloudSyncConfig struct { - SyncThingID string `json:"syncThingID"` - STSvrStatus string `json:"-"` - STSvrIsInSync bool `json:"-"` - STLocStatus string `json:"-"` - STLocIsInSync bool `json:"-"` -} - -// XdsEventRegisterArgs arguments used to register to XDS server events -type XdsEventRegisterArgs struct { - Name string `json:"name"` - ProjectID string `json:"filterProjectID"` -} - -// XdsEventFolderChange Folder change event structure -type XdsEventFolderChange struct { - Time string `json:"time"` - Type string `json:"type"` - Folder XdsFolderConfig `json:"folder"` + cmdList map[string]interface{} } // EventCB Event emitter callback @@ -145,16 +90,15 @@ func NewXdsServer(ctx *Context, conf xdsconfig.XDSServerConf) *XdsServer { sockEvents: make(map[string][]*caller), sockEventsLock: &sync.Mutex{}, logOut: ctx.Log.Out, + cmdList: make(map[string]interface{}), } } // Close Free and close XDS Server connection func (xs *XdsServer) Close() error { - xs.Connected = false + err := xs._Disconnected() xs.Disabled = true - xs.ioSock = nil - xs._NotifyState() - return nil + return err } // Connect Establish HTTP connection with XDS Server @@ -179,7 +123,7 @@ func (xs *XdsServer) Connect() error { time.Sleep(time.Second) } if retry == 0 { - // FIXME: re-use _reconnect to wait longer in background + // FIXME: re-use _Reconnect to wait longer in background return fmt.Errorf("Connection to XDS Server failure") } if err != nil { @@ -187,7 +131,7 @@ func (xs *XdsServer) Connect() error { } // Check HTTP connection and establish WS connection - err = xs._connect(false) + err = xs._Connect(false) return err } @@ -203,12 +147,12 @@ func (xs *XdsServer) SetLoggerOutput(out io.Writer) { } // SendCommand Send a command to XDS Server -func (xs *XdsServer) SendCommand(cmd string, body []byte) (*http.Response, error) { +func (xs *XdsServer) SendCommand(cmd string, body []byte, res interface{}) error { url := cmd if !strings.HasPrefix("/", cmd) { url = "/" + cmd } - return xs.client.HTTPPostWithRes(url, string(body)) + return xs.client.Post(url, string(body), res) } // GetVersion Send Get request to retrieve XDS Server version @@ -217,12 +161,12 @@ func (xs *XdsServer) GetVersion(res interface{}) error { } // GetFolders Send GET request to get current folder configuration -func (xs *XdsServer) GetFolders(folders *[]XdsFolderConfig) error { +func (xs *XdsServer) GetFolders(folders *[]xsapiv1.FolderConfig) error { return xs.client.Get("/folders", folders) } // FolderAdd Send POST request to add a folder -func (xs *XdsServer) FolderAdd(fld *XdsFolderConfig, res interface{}) error { +func (xs *XdsServer) FolderAdd(fld *xsapiv1.FolderConfig, res interface{}) error { err := xs.client.Post("/folders", fld, res) if err != nil { return fmt.Errorf("FolderAdd error: %s", err.Error()) @@ -241,10 +185,20 @@ func (xs *XdsServer) FolderSync(id string) error { } // FolderUpdate Send PUT request to update a folder -func (xs *XdsServer) FolderUpdate(fld *XdsFolderConfig, resFld *XdsFolderConfig) error { +func (xs *XdsServer) FolderUpdate(fld *xsapiv1.FolderConfig, resFld *xsapiv1.FolderConfig) error { return xs.client.Put("/folders/"+fld.ID, fld, resFld) } +// CommandExec Send POST request to execute a command +func (xs *XdsServer) CommandExec(args *xsapiv1.ExecArgs, res *xsapiv1.ExecResult) error { + return xs.client.Post("/exec", args, res) +} + +// CommandSignal Send POST request to send a signal to a command +func (xs *XdsServer) CommandSignal(args *xsapiv1.ExecSignalArgs, res *xsapiv1.ExecSigResult) error { + return xs.client.Post("/signal", args, res) +} + // SetAPIRouterGroup . func (xs *XdsServer) SetAPIRouterGroup(r *gin.RouterGroup) { xs.apiRouter = r @@ -267,8 +221,7 @@ func (xs *XdsServer) PassthroughGet(url string) { // Send Get request if err := xs.client.Get(nURL, &data); err != nil { if strings.Contains(err.Error(), "connection refused") { - xs.Connected = false - xs._NotifyState() + xs._Disconnected() } common.APIError(c, err.Error()) return @@ -325,13 +278,22 @@ func (xs *XdsServer) PassthroughPost(url string) { func (xs *XdsServer) EventRegister(evName string, id string) error { return xs.client.Post( "/events/register", - XdsEventRegisterArgs{ + xsapiv1.EventRegisterArgs{ Name: evName, ProjectID: id, }, nil) } +// EventEmit Emit a event to XDS Server through WS +func (xs *XdsServer) EventEmit(message string, args ...interface{}) error { + if xs.ioSock == nil { + return fmt.Errorf("Io.Socket not initialized") + } + + return xs.ioSock.Emit(message, args...) +} + // EventOn Register a callback on events reception func (xs *XdsServer) EventOn(evName string, privData interface{}, f EventCB) (uuid.UUID, error) { if xs.ioSock == nil { @@ -345,10 +307,7 @@ func (xs *XdsServer) EventOn(evName string, privData interface{}, f EventCB) (uu // Register listener only the first time evn := evName - // FIXME: use generic type: data interface{} instead of data XdsEventFolderChange - var err error - if evName == "event:folder-state-change" { - err = xs.ioSock.On(evn, func(data XdsEventFolderChange) error { + err := xs.ioSock.On(evn, func(data interface{}) error { xs.sockEventsLock.Lock() sEvts := make([]*caller, len(xs.sockEvents[evn])) copy(sEvts, xs.sockEvents[evn]) @@ -358,18 +317,6 @@ func (xs *XdsServer) EventOn(evName string, privData interface{}, f EventCB) (uu } return nil }) - } else { - err = xs.ioSock.On(evn, func(data interface{}) error { - xs.sockEventsLock.Lock() - sEvts := make([]*caller, len(xs.sockEvents[evn])) - copy(sEvts, xs.sockEvents[evn]) - xs.sockEventsLock.Unlock() - for _, c := range sEvts { - c.Func(c.PrivateData, data) - } - return nil - }) - } if err != nil { return uuid.Nil, err } @@ -383,6 +330,7 @@ func (xs *XdsServer) EventOn(evName string, privData interface{}, f EventCB) (uu } xs.sockEvents[evName] = append(xs.sockEvents[evName], c) + xs.LogSillyf("XS EventOn: sockEvents[\"%s\"]: len %d", evName, len(xs.sockEvents[evName])) return c.id, nil } @@ -404,29 +352,30 @@ func (xs *XdsServer) EventOff(evName string, id uuid.UUID) error { } } } + xs.LogSillyf("XS EventOff: sockEvents[\"%s\"]: len %d", evName, len(xs.sockEvents[evName])) return nil } // ProjectToFolder Convert Project structure to Folder structure -func (xs *XdsServer) ProjectToFolder(pPrj apiv1.ProjectConfig) *XdsFolderConfig { +func (xs *XdsServer) ProjectToFolder(pPrj xaapiv1.ProjectConfig) *xsapiv1.FolderConfig { stID := "" - if pPrj.Type == XdsTypeCloudSync { + if pPrj.Type == xsapiv1.TypeCloudSync { stID, _ = xs.SThg.IDGet() } // TODO: limit ClientData size and gzip it (see https://golang.org/pkg/compress/gzip/) - fPrj := XdsFolderConfig{ + fPrj := xsapiv1.FolderConfig{ ID: pPrj.ID, Label: pPrj.Label, ClientPath: pPrj.ClientPath, - Type: XdsFolderType(pPrj.Type), + Type: xsapiv1.FolderType(pPrj.Type), Status: pPrj.Status, IsInSync: pPrj.IsInSync, DefaultSdk: pPrj.DefaultSdk, ClientData: pPrj.ClientData, - DataPathMap: XdsPathMapConfig{ + DataPathMap: xsapiv1.PathMapConfig{ ServerPath: pPrj.ServerPath, }, - DataCloudSync: XdsCloudSyncConfig{ + DataCloudSync: xsapiv1.CloudSyncConfig{ SyncThingID: stID, STLocIsInSync: pPrj.IsInSync, STLocStatus: pPrj.Status, @@ -439,36 +388,36 @@ func (xs *XdsServer) ProjectToFolder(pPrj apiv1.ProjectConfig) *XdsFolderConfig } // FolderToProject Convert Folder structure to Project structure -func (xs *XdsServer) FolderToProject(fPrj XdsFolderConfig) apiv1.ProjectConfig { +func (xs *XdsServer) FolderToProject(fPrj xsapiv1.FolderConfig) xaapiv1.ProjectConfig { inSync := fPrj.IsInSync sts := fPrj.Status - if fPrj.Type == XdsTypeCloudSync { + if fPrj.Type == xsapiv1.TypeCloudSync { inSync = fPrj.DataCloudSync.STSvrIsInSync && fPrj.DataCloudSync.STLocIsInSync sts = fPrj.DataCloudSync.STSvrStatus switch fPrj.DataCloudSync.STLocStatus { - case apiv1.StatusErrorConfig, apiv1.StatusDisable, apiv1.StatusPause: + case xaapiv1.StatusErrorConfig, xaapiv1.StatusDisable, xaapiv1.StatusPause: sts = fPrj.DataCloudSync.STLocStatus break - case apiv1.StatusSyncing: - if sts != apiv1.StatusErrorConfig && sts != apiv1.StatusDisable && sts != apiv1.StatusPause { - sts = apiv1.StatusSyncing + case xaapiv1.StatusSyncing: + if sts != xaapiv1.StatusErrorConfig && sts != xaapiv1.StatusDisable && sts != xaapiv1.StatusPause { + sts = xaapiv1.StatusSyncing } break - case apiv1.StatusEnable: + case xaapiv1.StatusEnable: // keep STSvrStatus break } } - pPrj := apiv1.ProjectConfig{ + pPrj := xaapiv1.ProjectConfig{ ID: fPrj.ID, ServerID: xs.ID, Label: fPrj.Label, ClientPath: fPrj.ClientPath, ServerPath: fPrj.DataPathMap.ServerPath, - Type: apiv1.ProjectType(fPrj.Type), + Type: xaapiv1.ProjectType(fPrj.Type), Status: sts, IsInSync: inSync, DefaultSdk: fPrj.DefaultSdk, @@ -477,6 +426,33 @@ func (xs *XdsServer) FolderToProject(fPrj XdsFolderConfig) apiv1.ProjectConfig { return pPrj } +// CommandAdd Add a new command to the list of running commands +func (xs *XdsServer) CommandAdd(cmdID string, data interface{}) error { + if xs.CommandGet(cmdID) != nil { + return fmt.Errorf("command id already exist") + } + xs.cmdList[cmdID] = data + return nil +} + +// CommandDelete Delete a command from the command list +func (xs *XdsServer) CommandDelete(cmdID string) error { + if xs.CommandGet(cmdID) == nil { + return fmt.Errorf("unknown command id") + } + delete(xs.cmdList, cmdID) + return nil +} + +// CommandGet Retrieve a command data +func (xs *XdsServer) CommandGet(cmdID string) interface{} { + d, exist := xs.cmdList[cmdID] + if exist { + return d + } + return nil +} + /*** ** Private functions ***/ @@ -510,9 +486,9 @@ func (xs *XdsServer) _CreateConnectHTTP() error { return nil } -// Re-established connection -func (xs *XdsServer) _reconnect() error { - err := xs._connect(true) +// _Reconnect Re-established connection +func (xs *XdsServer) _Reconnect() error { + err := xs._Connect(true) if err == nil { // Reload projects list for this server err = xs.projects.Init(xs) @@ -520,10 +496,10 @@ func (xs *XdsServer) _reconnect() error { return err } -// Established HTTP and WS connection and retrieve XDSServer config -func (xs *XdsServer) _connect(reConn bool) error { +// _Connect Established HTTP and WS connection and retrieve XDSServer config +func (xs *XdsServer) _Connect(reConn bool) error { - xdsCfg := XdsServerConfig{} + xdsCfg := xsapiv1.APIConfig{} if err := xs.client.Get("/config", &xdsCfg); err != nil { xs.Connected = false if !reConn { @@ -532,18 +508,17 @@ func (xs *XdsServer) _connect(reConn bool) error { return err } - if reConn && xs.ID != xdsCfg.ID { - xs.Log.Warningf("Reconnected to server but ID differs: old=%s, new=%s", xs.ID, xdsCfg.ID) + if reConn && xs.ID != xdsCfg.ServerUID { + xs.Log.Warningf("Reconnected to server but ID differs: old=%s, new=%s", xs.ID, xdsCfg.ServerUID) } // Update local XDS config - xs.ID = xdsCfg.ID + xs.ID = xdsCfg.ServerUID xs.ServerConfig = &xdsCfg // Establish WS connection and register listen if err := xs._SocketConnect(); err != nil { - xs.Connected = false - xs._NotifyState() + xs._Disconnected() return err } @@ -552,7 +527,7 @@ func (xs *XdsServer) _connect(reConn bool) error { return nil } -// Create WebSocket (io.socket) connection +// _SocketConnect Create WebSocket (io.socket) connection func (xs *XdsServer) _SocketConnect() error { xs.Log.Infof("Connecting IO.socket for server %s (url %s)", xs.ID, xs.BaseURL) @@ -583,8 +558,7 @@ func (xs *XdsServer) _SocketConnect() error { if xs.CBOnDisconnect != nil { xs.CBOnDisconnect(err) } - xs.Connected = false - xs._NotifyState() + xs._Disconnected() // Try to reconnect during 15min (or at least while not disabled) go func() { @@ -602,7 +576,12 @@ func (xs *XdsServer) _SocketConnect() error { time.Sleep(time.Second * time.Duration(waitTime)) xs.Log.Infof("Try to reconnect to server %s (%d)", xs.BaseURL, count) - xs._reconnect() + err := xs._Reconnect() + if err != nil && + !(strings.Contains(err.Error(), "dial tcp") && strings.Contains(err.Error(), "connection refused")) { + xs.Log.Errorf("ERROR while reconnecting: %v", err.Error()) + } + } }() }) @@ -615,10 +594,22 @@ func (xs *XdsServer) _SocketConnect() error { return nil } -// Send event to notify changes +// _Disconnected Set XDS Server as disconnected +func (xs *XdsServer) _Disconnected() error { + // Clear all register events as socket is closed + for k := range xs.sockEvents { + delete(xs.sockEvents, k) + } + xs.Connected = false + xs.ioSock = nil + xs._NotifyState() + return nil +} + +// _NotifyState Send event to notify changes func (xs *XdsServer) _NotifyState() { - evSts := apiv1.ServerCfg{ + evSts := xaapiv1.ServerCfg{ ID: xs.ID, URL: xs.BaseURL, APIURL: xs.APIURL, @@ -626,7 +617,7 @@ func (xs *XdsServer) _NotifyState() { ConnRetry: xs.ConnRetry, Connected: xs.Connected, } - if err := xs.events.Emit(apiv1.EVTServerConfig, evSts, ""); err != nil { + if err := xs.events.Emit(xaapiv1.EVTServerConfig, evSts, ""); err != nil { xs.Log.Warningf("Cannot notify XdsServer state change: %v", err) } }