adapter/socketclient/socketclient.go

   1 // Copyright (c) 2019 Cisco and/or its affiliates.
   2 //
   3 // Licensed under the Apache License, Version 2.0 (the "License");
   4 // you may not use this file except in compliance with the License.
   5 // You may obtain a copy of the License at:
   6 //
   7 //     http://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 // Unless required by applicable law or agreed to in writing, software
  10 // distributed under the License is distributed on an "AS IS" BASIS,
  11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 // See the License for the specific language governing permissions and
  13 // limitations under the License.
  14
  15 package socketclient
  16
  17 import (
  18         "bufio"
  19         "encoding/binary"
  20         "errors"
  21         "fmt"
  22         "io"
  23         "net"
  24         "os"
  25         "path/filepath"
  26         "strings"
  27         "sync"
  28         "time"
  29
  30         "github.com/fsnotify/fsnotify"
  31         "github.com/sirupsen/logrus"
  32
  33         "git.fd.io/govpp.git/adapter"
  34         "git.fd.io/govpp.git/binapi/memclnt"
  35         "git.fd.io/govpp.git/codec"
  36 )
  37
  38 const (
  39         // DefaultSocketName is default VPP API socket file path.
  40         DefaultSocketName = "/run/vpp/api.sock"
  41         // DefaultClientName is used for identifying client in socket registration
  42         DefaultClientName = "govppsock"
  43 )
  44
  45 var (
  46
  47         // DefaultConnectTimeout is default timeout for connecting
  48         DefaultConnectTimeout = time.Second * 3
  49         // DefaultDisconnectTimeout is default timeout for discconnecting
  50         DefaultDisconnectTimeout = time.Millisecond * 100
  51         // MaxWaitReady defines maximum duration of waiting for socket file
  52         MaxWaitReady = time.Second * 10
  53 )
  54
  55 var (
  56         debug       = strings.Contains(os.Getenv("DEBUG_GOVPP"), "socketclient")
  57         debugMsgIds = strings.Contains(os.Getenv("DEBUG_GOVPP"), "msgtable")
  58
  59         logger = logrus.New()
  60         log    = logger.WithField("logger", "govpp/socketclient")
  61 )
  62
  63 // init initializes global logger
  64 func init() {
  65         if debug {
  66                 logger.Level = logrus.DebugLevel
  67                 log.Debug("govpp: debug level enabled for socketclient")
  68         }
  69 }
  70
  71 const socketMissing = `
  72 ------------------------------------------------------------
  73  No socket file found at: %s
  74  VPP binary API socket file is missing!
  75
  76   - is VPP running with socket for binapi enabled?
  77   - is the correct socket name configured?
  78
  79  To enable it add following section to your VPP config:
  80    socksvr {
  81      default
  82    }
  83 ------------------------------------------------------------
  84 `
  85
  86 var warnOnce sync.Once
  87
  88 func (c *socketClient) printMissingSocketMsg() {
  89         fmt.Fprintf(os.Stderr, socketMissing, c.sockAddr)
  90 }
  91
  92 type socketClient struct {
  93         sockAddr   string
  94         clientName string
  95
  96         conn   *net.UnixConn
  97         reader *bufio.Reader
  98         writer *bufio.Writer
  99
 100         connectTimeout    time.Duration
 101         disconnectTimeout time.Duration
 102
 103         msgCallback  adapter.MsgCallback
 104         clientIndex  uint32
 105         msgTable     map[string]uint16
 106         sockDelMsgId uint16
 107         writeMu      sync.Mutex
 108
 109         headerPool *sync.Pool
 110
 111         quit chan struct{}
 112         wg   sync.WaitGroup
 113 }
 114
 115 func NewVppClient(sockAddr string) *socketClient {
 116         if sockAddr == "" {
 117                 sockAddr = DefaultSocketName
 118         }
 119         return &socketClient{
 120                 sockAddr:          sockAddr,
 121                 clientName:        DefaultClientName,
 122                 connectTimeout:    DefaultConnectTimeout,
 123                 disconnectTimeout: DefaultDisconnectTimeout,
 124                 headerPool: &sync.Pool{New: func() interface{} {
 125                         return make([]byte, 16)
 126                 }},
 127                 msgCallback: func(msgID uint16, data []byte) {
 128                         log.Debugf("no callback set, dropping message: ID=%v len=%d", msgID, len(data))
 129                 },
 130         }
 131 }
 132
 133 // SetClientName sets a client name used for identification.
 134 func (c *socketClient) SetClientName(name string) {
 135         c.clientName = name
 136 }
 137
 138 // SetConnectTimeout sets timeout used during connecting.
 139 func (c *socketClient) SetConnectTimeout(t time.Duration) {
 140         c.connectTimeout = t
 141 }
 142
 143 // SetDisconnectTimeout sets timeout used during disconnecting.
 144 func (c *socketClient) SetDisconnectTimeout(t time.Duration) {
 145         c.disconnectTimeout = t
 146 }
 147
 148 func (c *socketClient) SetMsgCallback(cb adapter.MsgCallback) {
 149         log.Debug("SetMsgCallback")
 150         c.msgCallback = cb
 151 }
 152
 153 const legacySocketName = "/run/vpp-api.sock"
 154
 155 func (c *socketClient) checkLegacySocket() bool {
 156         if c.sockAddr == legacySocketName {
 157                 return false
 158         }
 159         log.Debugf("checking legacy socket: %s", legacySocketName)
 160         // check if socket exists
 161         if _, err := os.Stat(c.sockAddr); err == nil {
 162                 return false // socket exists
 163         } else if !os.IsNotExist(err) {
 164                 return false // some other error occurred
 165         }
 166         // check if legacy socket exists
 167         if _, err := os.Stat(legacySocketName); err == nil {
 168                 // legacy socket exists, update sockAddr
 169                 c.sockAddr = legacySocketName
 170                 return true
 171         }
 172         // no socket socket found
 173         return false
 174 }
 175
 176 // WaitReady checks socket file existence and waits for it if necessary
 177 func (c *socketClient) WaitReady() error {
 178         // check if socket already exists
 179         if _, err := os.Stat(c.sockAddr); err == nil {
 180                 return nil // socket exists, we are ready
 181         } else if !os.IsNotExist(err) {
 182                 return err // some other error occurred
 183         }
 184
 185         if c.checkLegacySocket() {
 186                 return nil
 187         }
 188
 189         // socket does not exist, watch for it
 190         watcher, err := fsnotify.NewWatcher()
 191         if err != nil {
 192                 return err
 193         }
 194         defer func() {
 195                 if err := watcher.Close(); err != nil {
 196                         log.Debugf("failed to close file watcher: %v", err)
 197                 }
 198         }()
 199
 200         // start directory watcher
 201         if err := watcher.Add(filepath.Dir(c.sockAddr)); err != nil {
 202                 return err
 203         }
 204
 205         timeout := time.NewTimer(MaxWaitReady)
 206         for {
 207                 select {
 208                 case <-timeout.C:
 209                         if c.checkLegacySocket() {
 210                                 return nil
 211                         }
 212                         return fmt.Errorf("timeout waiting (%s) for socket file: %s", MaxWaitReady, c.sockAddr)
 213
 214                 case e := <-watcher.Errors:
 215                         return e
 216
 217                 case ev := <-watcher.Events:
 218                         log.Debugf("watcher event: %+v", ev)
 219                         if ev.Name == c.sockAddr && (ev.Op&fsnotify.Create) == fsnotify.Create {
 220                                 // socket created, we are ready
 221                                 return nil
 222                         }
 223                 }
 224         }
 225 }
 226
 227 func (c *socketClient) Connect() error {
 228         c.checkLegacySocket()
 229
 230         // check if socket exists
 231         if _, err := os.Stat(c.sockAddr); os.IsNotExist(err) {
 232                 warnOnce.Do(c.printMissingSocketMsg)
 233                 return fmt.Errorf("VPP API socket file %s does not exist", c.sockAddr)
 234         } else if err != nil {
 235                 return fmt.Errorf("VPP API socket error: %v", err)
 236         }
 237
 238         if err := c.connect(c.sockAddr); err != nil {
 239                 return err
 240         }
 241
 242         if err := c.open(); err != nil {
 243                 _ = c.disconnect()
 244                 return err
 245         }
 246
 247         c.quit = make(chan struct{})
 248         c.wg.Add(1)
 249         go c.readerLoop()
 250
 251         return nil
 252 }
 253
 254 func (c *socketClient) Disconnect() error {
 255         if c.conn == nil {
 256                 return nil
 257         }
 258         log.Debugf("Disconnecting..")
 259
 260         close(c.quit)
 261
 262         if err := c.conn.CloseRead(); err != nil {
 263                 log.Debugf("closing readMsg failed: %v", err)
 264         }
 265
 266         // wait for readerLoop to return
 267         c.wg.Wait()
 268
 269         // Don't bother sending a vl_api_sockclnt_delete_t message,
 270         // just close the socket.
 271
 272         if err := c.disconnect(); err != nil {
 273                 return err
 274         }
 275
 276         return nil
 277 }
 278
 279 const defaultBufferSize = 4096
 280
 281 func (c *socketClient) connect(sockAddr string) error {
 282         addr := &net.UnixAddr{Name: sockAddr, Net: "unix"}
 283
 284         log.Debugf("Connecting to: %v", c.sockAddr)
 285
 286         conn, err := net.DialUnix("unix", nil, addr)
 287         if err != nil {
 288                 // we try different type of socket for backwards compatbility with VPP<=19.04
 289                 if strings.Contains(err.Error(), "wrong type for socket") {
 290                         addr.Net = "unixpacket"
 291                         log.Debugf("%s, retrying connect with type unixpacket", err)
 292                         conn, err = net.DialUnix("unixpacket", nil, addr)
 293                 }
 294                 if err != nil {
 295                         log.Debugf("Connecting to socket %s failed: %s", addr, err)
 296                         return err
 297                 }
 298         }
 299
 300         c.conn = conn
 301         log.Debugf("Connected to socket (local addr: %v)", c.conn.LocalAddr().(*net.UnixAddr))
 302
 303         c.reader = bufio.NewReaderSize(c.conn, defaultBufferSize)
 304         c.writer = bufio.NewWriterSize(c.conn, defaultBufferSize)
 305
 306         return nil
 307 }
 308
 309 func (c *socketClient) disconnect() error {
 310         log.Debugf("Closing socket")
 311         if err := c.conn.Close(); err != nil {
 312                 log.Debugln("Closing socket failed:", err)
 313                 return err
 314         }
 315         return nil
 316 }
 317
 318 const (
 319         sockCreateMsgId  = 15 // hard-coded sockclnt_create message ID
 320         createMsgContext = byte(123)
 321         deleteMsgContext = byte(124)
 322 )
 323
 324 func (c *socketClient) open() error {
 325         var msgCodec = codec.DefaultCodec
 326
 327         // Request socket client create
 328         req := &memclnt.SockclntCreate{
 329                 Name: c.clientName,
 330         }
 331         msg, err := msgCodec.EncodeMsg(req, sockCreateMsgId)
 332         if err != nil {
 333                 log.Debugln("Encode  error:", err)
 334                 return err
 335         }
 336         // set non-0 context
 337         msg[5] = createMsgContext
 338
 339         if err := c.writeMsg(msg); err != nil {
 340                 log.Debugln("Write error: ", err)
 341                 return err
 342         }
 343         msgReply, err := c.readMsgTimeout(nil, c.connectTimeout)
 344         if err != nil {
 345                 log.Println("Read error:", err)
 346                 return err
 347         }
 348
 349         reply := new(memclnt.SockclntCreateReply)
 350         if err := msgCodec.DecodeMsg(msgReply, reply); err != nil {
 351                 log.Println("Decoding sockclnt_create_reply failed:", err)
 352                 return err
 353         } else if reply.Response != 0 {
 354                 return fmt.Errorf("sockclnt_create_reply: response error (%d)", reply.Response)
 355         }
 356
 357         log.Debugf("SockclntCreateReply: Response=%v Index=%v Count=%v",
 358                 reply.Response, reply.Index, reply.Count)
 359
 360         c.clientIndex = reply.Index
 361         c.msgTable = make(map[string]uint16, reply.Count)
 362         for _, x := range reply.MessageTable {
 363                 msgName := strings.Split(x.Name, "\x00")[0]
 364                 name := strings.TrimSuffix(msgName, "\x13")
 365                 c.msgTable[name] = x.Index
 366                 if strings.HasPrefix(name, "sockclnt_delete_") {
 367                         c.sockDelMsgId = x.Index
 368                 }
 369                 if debugMsgIds {
 370                         log.Debugf(" - %4d: %q", x.Index, name)
 371                 }
 372         }
 373
 374         return nil
 375 }
 376
 377 func (c *socketClient) GetMsgID(msgName string, msgCrc string) (uint16, error) {
 378         if msgID, ok := c.msgTable[msgName+"_"+msgCrc]; ok {
 379                 return msgID, nil
 380         }
 381         return 0, &adapter.UnknownMsgError{
 382                 MsgName: msgName,
 383                 MsgCrc:  msgCrc,
 384         }
 385 }
 386
 387 func (c *socketClient) SendMsg(context uint32, data []byte) error {
 388         if len(data) < 10 {
 389                 return fmt.Errorf("invalid message data, length must be at least 10 bytes")
 390         }
 391         setMsgRequestHeader(data, c.clientIndex, context)
 392
 393         if debug {
 394                 log.Debugf("sendMsg (%d) context=%v client=%d: % 02X", len(data), context, c.clientIndex, data)
 395         }
 396
 397         if err := c.writeMsg(data); err != nil {
 398                 log.Debugln("writeMsg error: ", err)
 399                 return err
 400         }
 401
 402         return nil
 403 }
 404
 405 // setMsgRequestHeader sets client index and context in the message request header
 406 //
 407 // Message request has following structure:
 408 //
 409 //    type msgRequestHeader struct {
 410 //        MsgID       uint16
 411 //        ClientIndex uint32
 412 //        Context     uint32
 413 //    }
 414 //
 415 func setMsgRequestHeader(data []byte, clientIndex, context uint32) {
 416         // message ID is already set
 417         binary.BigEndian.PutUint32(data[2:6], clientIndex)
 418         binary.BigEndian.PutUint32(data[6:10], context)
 419 }
 420
 421 func (c *socketClient) writeMsg(msg []byte) error {
 422         // we lock to prevent mixing multiple message writes
 423         c.writeMu.Lock()
 424         defer c.writeMu.Unlock()
 425
 426         header := c.headerPool.Get().([]byte)
 427         err := writeMsgHeader(c.writer, header, len(msg))
 428         if err != nil {
 429                 return err
 430         }
 431         c.headerPool.Put(header)
 432
 433         if err := writeMsgData(c.writer, msg, c.writer.Size()); err != nil {
 434                 return err
 435         }
 436
 437         if err := c.writer.Flush(); err != nil {
 438                 return err
 439         }
 440
 441         log.Debugf(" -- writeMsg done")
 442
 443         return nil
 444 }
 445
 446 func writeMsgHeader(w io.Writer, header []byte, dataLen int) error {
 447         binary.BigEndian.PutUint32(header[8:12], uint32(dataLen))
 448
 449         n, err := w.Write(header)
 450         if err != nil {
 451                 return err
 452         }
 453         if debug {
 454                 log.Debugf(" - header sent (%d/%d): % 0X", n, len(header), header)
 455         }
 456
 457         return nil
 458 }
 459
 460 func writeMsgData(w io.Writer, msg []byte, writerSize int) error {
 461         for i := 0; i <= len(msg)/writerSize; i++ {
 462                 x := i*writerSize + writerSize
 463                 if x > len(msg) {
 464                         x = len(msg)
 465                 }
 466                 if debug {
 467                         log.Debugf(" - x=%v i=%v len=%v mod=%v", x, i, len(msg), len(msg)/writerSize)
 468                 }
 469                 n, err := w.Write(msg[i*writerSize : x])
 470                 if err != nil {
 471                         return err
 472                 }
 473                 if debug {
 474                         log.Debugf(" - data sent x=%d (%d/%d): % 0X", x, n, len(msg), msg)
 475                 }
 476         }
 477         return nil
 478 }
 479
 480 func (c *socketClient) readerLoop() {
 481         defer c.wg.Done()
 482         defer log.Debugf("reader loop done")
 483
 484         var buf [8192]byte
 485
 486         for {
 487                 select {
 488                 case <-c.quit:
 489                         return
 490                 default:
 491                 }
 492
 493                 msg, err := c.readMsg(buf[:])
 494                 if err != nil {
 495                         if isClosedError(err) {
 496                                 return
 497                         }
 498                         log.Debugf("readMsg error: %v", err)
 499                         continue
 500                 }
 501
 502                 msgID, context := getMsgReplyHeader(msg)
 503                 if debug {
 504                         log.Debugf("recvMsg (%d) msgID=%d context=%v", len(msg), msgID, context)
 505                 }
 506
 507                 c.msgCallback(msgID, msg)
 508         }
 509 }
 510
 511 // getMsgReplyHeader gets message ID and context from the message reply header
 512 //
 513 // Message reply has following structure:
 514 //
 515 //    type msgReplyHeader struct {
 516 //        MsgID       uint16
 517 //        Context     uint32
 518 //    }
 519 //
 520 func getMsgReplyHeader(msg []byte) (msgID uint16, context uint32) {
 521         msgID = binary.BigEndian.Uint16(msg[0:2])
 522         context = binary.BigEndian.Uint32(msg[2:6])
 523         return
 524 }
 525
 526 func (c *socketClient) readMsgTimeout(buf []byte, timeout time.Duration) ([]byte, error) {
 527         // set read deadline
 528         readDeadline := time.Now().Add(timeout)
 529         if err := c.conn.SetReadDeadline(readDeadline); err != nil {
 530                 return nil, err
 531         }
 532
 533         // read message
 534         msgReply, err := c.readMsg(buf)
 535         if err != nil {
 536                 return nil, err
 537         }
 538
 539         // reset read deadline
 540         if err := c.conn.SetReadDeadline(time.Time{}); err != nil {
 541                 return nil, err
 542         }
 543
 544         return msgReply, nil
 545 }
 546
 547 func (c *socketClient) readMsg(buf []byte) ([]byte, error) {
 548         log.Debug("reading msg..")
 549
 550         header := c.headerPool.Get().([]byte)
 551         msgLen, err := readMsgHeader(c.reader, header)
 552         if err != nil {
 553                 return nil, err
 554         }
 555         c.headerPool.Put(header)
 556
 557         msg, err := readMsgData(c.reader, buf, msgLen)
 558
 559         log.Debugf(" -- readMsg done (buffered: %d)", c.reader.Buffered())
 560
 561         return msg, nil
 562 }
 563
 564 func readMsgHeader(r io.Reader, header []byte) (int, error) {
 565         n, err := io.ReadAtLeast(r, header, 16)
 566         if err != nil {
 567                 return 0, err
 568         }
 569         if n == 0 {
 570                 log.Debugln("zero bytes header")
 571                 return 0, nil
 572         } else if n != 16 {
 573                 log.Debugf("invalid header (%d bytes): % 0X", n, header[:n])
 574                 return 0, fmt.Errorf("invalid header (expected 16 bytes, got %d)", n)
 575         }
 576
 577         dataLen := binary.BigEndian.Uint32(header[8:12])
 578
 579         return int(dataLen), nil
 580 }
 581
 582 func readMsgData(r io.Reader, buf []byte, dataLen int) ([]byte, error) {
 583         var msg []byte
 584         if buf == nil || len(buf) < dataLen {
 585                 msg = make([]byte, dataLen)
 586         } else {
 587                 msg = buf[0:dataLen]
 588         }
 589
 590         n, err := r.Read(msg)
 591         if err != nil {
 592                 return nil, err
 593         }
 594         if debug {
 595                 log.Debugf(" - read data (%d bytes): % 0X", n, msg[:n])
 596         }
 597
 598         if dataLen > n {
 599                 remain := dataLen - n
 600                 log.Debugf("continue reading remaining %d bytes", remain)
 601                 view := msg[n:]
 602
 603                 for remain > 0 {
 604                         nbytes, err := r.Read(view)
 605                         if err != nil {
 606                                 return nil, err
 607                         } else if nbytes == 0 {
 608                                 return nil, fmt.Errorf("zero nbytes")
 609                         }
 610
 611                         remain -= nbytes
 612                         log.Debugf("another data received: %d bytes (remain: %d)", nbytes, remain)
 613
 614                         view = view[nbytes:]
 615                 }
 616         }
 617
 618         return msg, nil
 619 }
 620
 621 func isClosedError(err error) bool {
 622         if errors.Is(err, io.EOF) {
 623                 return true
 624         }
 625         return strings.HasSuffix(err.Error(), "use of closed network connection")
 626 }