X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=adapter%2Fsocketclient%2Fsocketclient.go;h=043d253ed13094db5d327fcf83930a2ed0a3e1a5;hb=ccb7b913d54fafdf08b36ac7eb36e462b1ecc9eb;hp=2b676454617e6875ad2ae45d4e5892311d02f368;hpb=87e79ec9fa48e5f8b2eb949c337488db75a17b29;p=govpp.git diff --git a/adapter/socketclient/socketclient.go b/adapter/socketclient/socketclient.go index 2b67645..043d253 100644 --- a/adapter/socketclient/socketclient.go +++ b/adapter/socketclient/socketclient.go @@ -1,3 +1,17 @@ +// Copyright (c) 2019 Cisco and/or its affiliates. +// +// 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 socketclient import ( @@ -18,22 +32,22 @@ import ( "git.fd.io/govpp.git/adapter" "git.fd.io/govpp.git/codec" - "git.fd.io/govpp.git/examples/bin_api/memclnt" ) const ( - // DefaultSocketName is default VPP API socket file name - DefaultSocketName = "/run/vpp-api.sock" + // DefaultSocketName is default VPP API socket file path. + DefaultSocketName = adapter.DefaultBinapiSocket + legacySocketName = "/run/vpp-api.sock" ) var ( // DefaultConnectTimeout is default timeout for connecting DefaultConnectTimeout = time.Second * 3 // DefaultDisconnectTimeout is default timeout for discconnecting - DefaultDisconnectTimeout = time.Second + DefaultDisconnectTimeout = time.Millisecond * 100 // MaxWaitReady defines maximum duration before waiting for socket file // times out - MaxWaitReady = time.Second * 15 + MaxWaitReady = time.Second * 10 // ClientName is used for identifying client in socket registration ClientName = "govppsock" ) @@ -44,7 +58,8 @@ var ( // DebugMsgIds is global variable that determines debug mode for msg ids DebugMsgIds = os.Getenv("DEBUG_GOVPP_SOCKMSG") != "" - Log = logger.New() // global logger + // Log is global logger + Log = logger.New() ) // init initializes global logger, which logs debug level messages to stdout. @@ -52,14 +67,37 @@ func init() { Log.Out = os.Stdout if Debug { Log.Level = logger.DebugLevel + Log.Debug("govpp/socketclient: enabled debug mode") } } +const socketMissing = ` +------------------------------------------------------------ + No socket file found at: %s + VPP binary API socket file is missing! + + - is VPP running with socket for binapi enabled? + - is the correct socket name configured? + + To enable it add following section to your VPP config: + socksvr { + default + } +------------------------------------------------------------ +` + +var warnOnce sync.Once + +func (c *vppClient) printMissingSocketMsg() { + fmt.Fprintf(os.Stderr, socketMissing, c.sockAddr) +} + type vppClient struct { sockAddr string - conn *net.UnixConn - reader *bufio.Reader - writer *bufio.Writer + + conn *net.UnixConn + reader *bufio.Reader + writer *bufio.Writer connectTimeout time.Duration disconnectTimeout time.Duration @@ -98,62 +136,100 @@ func (c *vppClient) SetDisconnectTimeout(t time.Duration) { c.disconnectTimeout = t } +func (c *vppClient) SetMsgCallback(cb adapter.MsgCallback) { + Log.Debug("SetMsgCallback") + c.cb = cb +} + +func (c *vppClient) checkLegacySocket() bool { + if c.sockAddr == legacySocketName { + return false + } + Log.Debugf("checking legacy socket: %s", legacySocketName) + // check if socket exists + if _, err := os.Stat(c.sockAddr); err == nil { + return false // socket exists + } else if !os.IsNotExist(err) { + return false // some other error occurred + } + // check if legacy socket exists + if _, err := os.Stat(legacySocketName); err == nil { + // legacy socket exists, update sockAddr + c.sockAddr = legacySocketName + return true + } + // no socket socket found + return false +} + // WaitReady checks socket file existence and waits for it if necessary func (c *vppClient) WaitReady() error { - // check if file at the path already exists + // check if socket already exists if _, err := os.Stat(c.sockAddr); err == nil { + return nil // socket exists, we are ready + } else if !os.IsNotExist(err) { + return err // some other error occurred + } + + if c.checkLegacySocket() { return nil - } else if os.IsExist(err) { - return err } - // if not, watch for it + // socket does not exist, watch for it watcher, err := fsnotify.NewWatcher() if err != nil { return err } defer func() { if err := watcher.Close(); err != nil { - Log.Errorf("failed to close file watcher: %v", err) + Log.Warnf("failed to close file watcher: %v", err) } }() - // start watching directory + // start directory watcher if err := watcher.Add(filepath.Dir(c.sockAddr)); err != nil { return err } + timeout := time.NewTimer(MaxWaitReady) for { select { - case <-time.After(MaxWaitReady): - return fmt.Errorf("waiting for socket file timed out (%s)", MaxWaitReady) + case <-timeout.C: + if c.checkLegacySocket() { + return nil + } + return fmt.Errorf("timeout waiting (%s) for socket file: %s", MaxWaitReady, c.sockAddr) + case e := <-watcher.Errors: return e + case ev := <-watcher.Events: Log.Debugf("watcher event: %+v", ev) - if ev.Name == c.sockAddr { - if (ev.Op & fsnotify.Create) == fsnotify.Create { - // socket was created, we are ready - return nil - } + if ev.Name == c.sockAddr && (ev.Op&fsnotify.Create) == fsnotify.Create { + // socket created, we are ready + return nil } } } } -func (c *vppClient) SetMsgCallback(cb adapter.MsgCallback) { - Log.Debug("SetMsgCallback") - c.cb = cb -} - func (c *vppClient) Connect() error { - Log.Debugf("Connecting to: %v", c.sockAddr) + c.checkLegacySocket() + + // check if socket exists + if _, err := os.Stat(c.sockAddr); os.IsNotExist(err) { + warnOnce.Do(c.printMissingSocketMsg) + return fmt.Errorf("VPP API socket file %s does not exist", c.sockAddr) + } else if err != nil { + return fmt.Errorf("VPP API socket error: %v", err) + } if err := c.connect(c.sockAddr); err != nil { return err } if err := c.open(); err != nil { + c.disconnect() return err } @@ -164,30 +240,66 @@ func (c *vppClient) Connect() error { return nil } -func (c *vppClient) connect(sockAddr string) error { - addr, err := net.ResolveUnixAddr("unixpacket", sockAddr) - if err != nil { - Log.Debugln("ResolveUnixAddr error:", err) +func (c *vppClient) Disconnect() error { + if c.conn == nil { + return nil + } + Log.Debugf("Disconnecting..") + + close(c.quit) + + if err := c.conn.CloseRead(); err != nil { + Log.Debugf("closing read failed: %v", err) + } + + // wait for readerLoop to return + c.wg.Wait() + + if err := c.close(); err != nil { + Log.Debugf("closing failed: %v", err) + } + + if err := c.disconnect(); err != nil { return err } + return nil +} + +func (c *vppClient) connect(sockAddr string) error { + addr := &net.UnixAddr{Name: sockAddr, Net: "unix"} + + Log.Debugf("Connecting to: %v", c.sockAddr) + conn, err := net.DialUnix("unix", nil, addr) if err != nil { + // we try different type of socket for backwards compatbility with VPP<=19.04 if strings.Contains(err.Error(), "wrong type for socket") { + addr.Net = "unixpacket" + Log.Debugf("%s, retrying connect with type unixpacket", err) conn, err = net.DialUnix("unixpacket", nil, addr) } if err != nil { - Log.Debugln("Dial error:", err) + Log.Debugf("Connecting to socket %s failed: %s", addr, err) return err } } c.conn = conn + Log.Debugf("Connected to socket (local addr: %v)", c.conn.LocalAddr().(*net.UnixAddr)) + c.reader = bufio.NewReader(c.conn) c.writer = bufio.NewWriter(c.conn) - Log.Debugf("Connected to socket: %v", addr) + return nil +} +func (c *vppClient) disconnect() error { + Log.Debugf("Closing socket") + if err := c.conn.Close(); err != nil { + Log.Debugln("Closing socket failed:", err) + return err + } return nil } @@ -200,9 +312,7 @@ const ( func (c *vppClient) open() error { msgCodec := new(codec.MsgCodec) - req := &memclnt.SockclntCreate{ - Name: []byte(ClientName), - } + req := &SockclntCreate{Name: ClientName} msg, err := msgCodec.EncodeMsg(req, sockCreateMsgId) if err != nil { Log.Debugln("Encode error:", err) @@ -230,7 +340,7 @@ func (c *vppClient) open() error { return err } - reply := new(memclnt.SockclntCreateReply) + reply := new(SockclntCreateReply) if err := msgCodec.DecodeMsg(msgReply, reply); err != nil { Log.Println("Decode error:", err) return err @@ -242,7 +352,8 @@ func (c *vppClient) open() error { c.clientIndex = reply.Index c.msgTable = make(map[string]uint16, reply.Count) for _, x := range reply.MessageTable { - name := string(bytes.TrimSuffix(bytes.Split(x.Name, []byte{0x00})[0], []byte{0x13})) + msgName := strings.Split(x.Name, "\x00")[0] + name := strings.TrimSuffix(msgName, "\x13") c.msgTable[name] = x.Index if strings.HasPrefix(name, "sockclnt_delete_") { c.sockDelMsgId = x.Index @@ -255,38 +366,10 @@ func (c *vppClient) open() error { return nil } -func (c *vppClient) Disconnect() error { - if c.conn == nil { - return nil - } - Log.Debugf("Disconnecting..") - - close(c.quit) - - // force readerLoop to timeout - if err := c.conn.SetReadDeadline(time.Now()); err != nil { - return err - } - - // wait for readerLoop to return - c.wg.Wait() - - if err := c.close(); err != nil { - return err - } - - if err := c.conn.Close(); err != nil { - Log.Debugln("Close socket conn failed:", err) - return err - } - - return nil -} - func (c *vppClient) close() error { msgCodec := new(codec.MsgCodec) - req := &memclnt.SockclntDelete{ + req := &SockclntDelete{ Index: c.clientIndex, } msg, err := msgCodec.EncodeMsg(req, c.sockDelMsgId) @@ -297,7 +380,7 @@ func (c *vppClient) close() error { // set non-0 context msg[5] = deleteMsgContext - Log.Debugf("sending socklntDel (%d byes): % 0X\n", len(msg), msg) + Log.Debugf("sending socklntDel (%d byes): % 0X", len(msg), msg) if err := c.write(msg); err != nil { Log.Debugln("Write error: ", err) return err @@ -321,7 +404,7 @@ func (c *vppClient) close() error { return err } - reply := new(memclnt.SockclntDeleteReply) + reply := new(SockclntDeleteReply) if err := msgCodec.DecodeMsg(msgReply, reply); err != nil { Log.Debugln("Decode error:", err) return err @@ -358,7 +441,7 @@ func (c *vppClient) SendMsg(context uint32, data []byte) error { } copy(data[2:], buf.Bytes()) - Log.Debugf("SendMsg (%d) context=%v client=%d: data: % 02X", len(data), context, c.clientIndex, data) + Log.Debugf("sendMsg (%d) context=%v client=%d: data: % 02X", len(data), context, c.clientIndex, data) if err := c.write(data); err != nil { Log.Debugln("write error: ", err) @@ -388,26 +471,24 @@ func (c *vppClient) write(msg []byte) error { Log.Debugf(" - header sent (%d/%d): % 0X", n, len(header), header) } - if err := c.writer.Flush(); err != nil { - return err - } - - for i := 0; i <= len(msg)/c.writer.Size(); i++ { - x := i*c.writer.Size() + c.writer.Size() + writerSize := c.writer.Size() + for i := 0; i <= len(msg)/writerSize; i++ { + x := i*writerSize + writerSize if x > len(msg) { x = len(msg) } - Log.Debugf("x=%v i=%v len=%v mod=%v\n", x, i, len(msg), len(msg)/c.writer.Size()) - if n, err := c.writer.Write(msg[i*c.writer.Size() : x]); err != nil { + Log.Debugf(" - x=%v i=%v len=%v mod=%v", x, i, len(msg), len(msg)/writerSize) + if n, err := c.writer.Write(msg[i*writerSize : x]); err != nil { return err } else { Log.Debugf(" - msg sent x=%d (%d/%d): % 0X", x, n, len(msg), msg) } - if err := c.writer.Flush(); err != nil { - return err - } - } + if err := c.writer.Flush(); err != nil { + return err + } + + Log.Debugf(" -- write done") return nil } @@ -419,10 +500,11 @@ type msgHeader struct { func (c *vppClient) readerLoop() { defer c.wg.Done() + defer Log.Debugf("reader quit") + for { select { case <-c.quit: - Log.Debugf("reader quit") return default: } @@ -432,9 +514,10 @@ func (c *vppClient) readerLoop() { if isClosedError(err) { return } - Log.Debugf("READ FAILED: %v", err) + Log.Debugf("read failed: %v", err) continue } + h := new(msgHeader) if err := struc.Unpack(bytes.NewReader(msg), h); err != nil { Log.Debugf("unpacking header failed: %v", err) @@ -453,22 +536,22 @@ type msgheader struct { } func (c *vppClient) read() ([]byte, error) { - Log.Debug("reading next msg..") + Log.Debug(" reading next msg..") header := make([]byte, 16) n, err := io.ReadAtLeast(c.reader, header, 16) if err != nil { return nil, err - } else if n == 0 { + } + if n == 0 { Log.Debugln("zero bytes header") return nil, nil - } - if n != 16 { - Log.Debug("invalid header data (%d): % 0X", n, header[:n]) + } else if n != 16 { + Log.Debugf("invalid header data (%d): % 0X", n, header[:n]) return nil, fmt.Errorf("invalid header (expected 16 bytes, got %d)", n) } - Log.Debugf(" - read header %d bytes: % 0X", n, header) + Log.Debugf(" read header %d bytes: % 0X", n, header) h := &msgheader{} if err := struc.Unpack(bytes.NewReader(header[:]), h); err != nil { @@ -483,7 +566,7 @@ func (c *vppClient) read() ([]byte, error) { if err != nil { return nil, err } - Log.Debugf(" - read msg %d bytes (%d buffered)", n, c.reader.Buffered()) + Log.Debugf(" - read msg %d bytes (%d buffered) % 0X", n, c.reader.Buffered(), msg[:n]) if msgLen > n { remain := msgLen - n @@ -491,7 +574,6 @@ func (c *vppClient) read() ([]byte, error) { view := msg[n:] for remain > 0 { - nbytes, err := c.reader.Read(view) if err != nil { return nil, err @@ -506,6 +588,8 @@ func (c *vppClient) read() ([]byte, error) { } } + Log.Debugf(" -- read done (buffered: %d)", c.reader.Buffered()) + return msg, nil }