Merge pull request #173 from bergwolf/proxy

runtime: add builtin proxy and shim capability

Gopkg.lock

@@ -77,6 +77,11 @@
   revision = "925541529c1fa6821df4e44ce2723319eb2be768"
   version = "v1.0.0"
 
+[[projects]]
+  name = "github.com/hashicorp/yamux"
+  packages = ["."]
+  revision = "f5742cb6b85602e7fa834e9d5d91a7d7fa850824"
+
 [[projects]]
   name = "github.com/intel/govmm"
   packages = ["qemu"]
@@ -88,7 +93,7 @@
     "protocols/client",
     "protocols/grpc"
   ]
-  revision = "a93071539feee29bfa22b6184380d3fd7c156ef7"
+  revision = "c8199f60759a1d52932df37a1af710cdeada8c81"
 
 [[projects]]
   name = "github.com/kubernetes-incubator/cri-o"
@@ -244,6 +249,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "f7c3a1b7f5cb5f891a3badcb7323f3b5fc0fa79f69dd6532ec2e2be2baafaf98"
+  inputs-digest = "2df221b473722077f7961415583a1e35c1e221f1ea2e2dbb8e98858cf42417b4"
   solver-name = "gps-cdcl"
   solver-version = 1

Gopkg.toml

@@ -60,7 +60,7 @@
 
 [[constraint]]
   name = "github.com/kata-containers/agent"
-  revision = "a93071539feee29bfa22b6184380d3fd7c156ef7"
+  revision = "c8199f60759a1d52932df37a1af710cdeada8c81"
 
 [[constraint]]
   name = "github.com/containerd/cri-containerd"
@@ -74,3 +74,7 @@
   non-go = true
   go-tests = true
   unused-packages = true
+
+[[constraint]]
+  branch = "master"
+  name = "github.com/hashicorp/yamux"

vendor/github.com/hashicorp/yamux/LICENSE

@@ -0,0 +1,362 @@
+Mozilla Public License, version 2.0
+
+1. Definitions
+
+1.1. "Contributor"
+
+     means each individual or legal entity that creates, contributes to the
+     creation of, or owns Covered Software.
+
+1.2. "Contributor Version"
+
+     means the combination of the Contributions of others (if any) used by a
+     Contributor and that particular Contributor's Contribution.
+
+1.3. "Contribution"
+
+     means Covered Software of a particular Contributor.
+
+1.4. "Covered Software"
+
+     means Source Code Form to which the initial Contributor has attached the
+     notice in Exhibit A, the Executable Form of such Source Code Form, and
+     Modifications of such Source Code Form, in each case including portions
+     thereof.
+
+1.5. "Incompatible With Secondary Licenses"
+     means
+
+     a. that the initial Contributor has attached the notice described in
+        Exhibit B to the Covered Software; or
+
+     b. that the Covered Software was made available under the terms of
+        version 1.1 or earlier of the License, but not also under the terms of
+        a Secondary License.
+
+1.6. "Executable Form"
+
+     means any form of the work other than Source Code Form.
+
+1.7. "Larger Work"
+
+     means a work that combines Covered Software with other material, in a
+     separate file or files, that is not Covered Software.
+
+1.8. "License"
+
+     means this document.
+
+1.9. "Licensable"
+
+     means having the right to grant, to the maximum extent possible, whether
+     at the time of the initial grant or subsequently, any and all of the
+     rights conveyed by this License.
+
+1.10. "Modifications"
+
+     means any of the following:
+
+     a. any file in Source Code Form that results from an addition to,
+        deletion from, or modification of the contents of Covered Software; or
+
+     b. any new file in Source Code Form that contains any Covered Software.
+
+1.11. "Patent Claims" of a Contributor
+
+      means any patent claim(s), including without limitation, method,
+      process, and apparatus claims, in any patent Licensable by such
+      Contributor that would be infringed, but for the grant of the License,
+      by the making, using, selling, offering for sale, having made, import,
+      or transfer of either its Contributions or its Contributor Version.
+
+1.12. "Secondary License"
+
+      means either the GNU General Public License, Version 2.0, the GNU Lesser
+      General Public License, Version 2.1, the GNU Affero General Public
+      License, Version 3.0, or any later versions of those licenses.
+
+1.13. "Source Code Form"
+
+      means the form of the work preferred for making modifications.
+
+1.14. "You" (or "Your")
+
+      means an individual or a legal entity exercising rights under this
+      License. For legal entities, "You" includes any entity that controls, is
+      controlled by, or is under common control with You. For purposes of this
+      definition, "control" means (a) the power, direct or indirect, to cause
+      the direction or management of such entity, whether by contract or
+      otherwise, or (b) ownership of more than fifty percent (50%) of the
+      outstanding shares or beneficial ownership of such entity.
+
+
+2. License Grants and Conditions
+
+2.1. Grants
+
+     Each Contributor hereby grants You a world-wide, royalty-free,
+     non-exclusive license:
+
+     a. under intellectual property rights (other than patent or trademark)
+        Licensable by such Contributor to use, reproduce, make available,
+        modify, display, perform, distribute, and otherwise exploit its
+        Contributions, either on an unmodified basis, with Modifications, or
+        as part of a Larger Work; and
+
+     b. under Patent Claims of such Contributor to make, use, sell, offer for
+        sale, have made, import, and otherwise transfer either its
+        Contributions or its Contributor Version.
+
+2.2. Effective Date
+
+     The licenses granted in Section 2.1 with respect to any Contribution
+     become effective for each Contribution on the date the Contributor first
+     distributes such Contribution.
+
+2.3. Limitations on Grant Scope
+
+     The licenses granted in this Section 2 are the only rights granted under
+     this License. No additional rights or licenses will be implied from the
+     distribution or licensing of Covered Software under this License.
+     Notwithstanding Section 2.1(b) above, no patent license is granted by a
+     Contributor:
+
+     a. for any code that a Contributor has removed from Covered Software; or
+
+     b. for infringements caused by: (i) Your and any other third party's
+        modifications of Covered Software, or (ii) the combination of its
+        Contributions with other software (except as part of its Contributor
+        Version); or
+
+     c. under Patent Claims infringed by Covered Software in the absence of
+        its Contributions.
+
+     This License does not grant any rights in the trademarks, service marks,
+     or logos of any Contributor (except as may be necessary to comply with
+     the notice requirements in Section 3.4).
+
+2.4. Subsequent Licenses
+
+     No Contributor makes additional grants as a result of Your choice to
+     distribute the Covered Software under a subsequent version of this
+     License (see Section 10.2) or under the terms of a Secondary License (if
+     permitted under the terms of Section 3.3).
+
+2.5. Representation
+
+     Each Contributor represents that the Contributor believes its
+     Contributions are its original creation(s) or it has sufficient rights to
+     grant the rights to its Contributions conveyed by this License.
+
+2.6. Fair Use
+
+     This License is not intended to limit any rights You have under
+     applicable copyright doctrines of fair use, fair dealing, or other
+     equivalents.
+
+2.7. Conditions
+
+     Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted in
+     Section 2.1.
+
+
+3. Responsibilities
+
+3.1. Distribution of Source Form
+
+     All distribution of Covered Software in Source Code Form, including any
+     Modifications that You create or to which You contribute, must be under
+     the terms of this License. You must inform recipients that the Source
+     Code Form of the Covered Software is governed by the terms of this
+     License, and how they can obtain a copy of this License. You may not
+     attempt to alter or restrict the recipients' rights in the Source Code
+     Form.
+
+3.2. Distribution of Executable Form
+
+     If You distribute Covered Software in Executable Form then:
+
+     a. such Covered Software must also be made available in Source Code Form,
+        as described in Section 3.1, and You must inform recipients of the
+        Executable Form how they can obtain a copy of such Source Code Form by
+        reasonable means in a timely manner, at a charge no more than the cost
+        of distribution to the recipient; and
+
+     b. You may distribute such Executable Form under the terms of this
+        License, or sublicense it under different terms, provided that the
+        license for the Executable Form does not attempt to limit or alter the
+        recipients' rights in the Source Code Form under this License.
+
+3.3. Distribution of a Larger Work
+
+     You may create and distribute a Larger Work under terms of Your choice,
+     provided that You also comply with the requirements of this License for
+     the Covered Software. If the Larger Work is a combination of Covered
+     Software with a work governed by one or more Secondary Licenses, and the
+     Covered Software is not Incompatible With Secondary Licenses, this
+     License permits You to additionally distribute such Covered Software
+     under the terms of such Secondary License(s), so that the recipient of
+     the Larger Work may, at their option, further distribute the Covered
+     Software under the terms of either this License or such Secondary
+     License(s).
+
+3.4. Notices
+
+     You may not remove or alter the substance of any license notices
+     (including copyright notices, patent notices, disclaimers of warranty, or
+     limitations of liability) contained within the Source Code Form of the
+     Covered Software, except that You may alter any license notices to the
+     extent required to remedy known factual inaccuracies.
+
+3.5. Application of Additional Terms
+
+     You may choose to offer, and to charge a fee for, warranty, support,
+     indemnity or liability obligations to one or more recipients of Covered
+     Software. However, You may do so only on Your own behalf, and not on
+     behalf of any Contributor. You must make it absolutely clear that any
+     such warranty, support, indemnity, or liability obligation is offered by
+     You alone, and You hereby agree to indemnify every Contributor for any
+     liability incurred by such Contributor as a result of warranty, support,
+     indemnity or liability terms You offer. You may include additional
+     disclaimers of warranty and limitations of liability specific to any
+     jurisdiction.
+
+4. Inability to Comply Due to Statute or Regulation
+
+   If it is impossible for You to comply with any of the terms of this License
+   with respect to some or all of the Covered Software due to statute,
+   judicial order, or regulation then You must: (a) comply with the terms of
+   this License to the maximum extent possible; and (b) describe the
+   limitations and the code they affect. Such description must be placed in a
+   text file included with all distributions of the Covered Software under
+   this License. Except to the extent prohibited by statute or regulation,
+   such description must be sufficiently detailed for a recipient of ordinary
+   skill to be able to understand it.
+
+5. Termination
+
+5.1. The rights granted under this License will terminate automatically if You
+     fail to comply with any of its terms. However, if You become compliant,
+     then the rights granted under this License from a particular Contributor
+     are reinstated (a) provisionally, unless and until such Contributor
+     explicitly and finally terminates Your grants, and (b) on an ongoing
+     basis, if such Contributor fails to notify You of the non-compliance by
+     some reasonable means prior to 60 days after You have come back into
+     compliance. Moreover, Your grants from a particular Contributor are
+     reinstated on an ongoing basis if such Contributor notifies You of the
+     non-compliance by some reasonable means, this is the first time You have
+     received notice of non-compliance with this License from such
+     Contributor, and You become compliant prior to 30 days after Your receipt
+     of the notice.
+
+5.2. If You initiate litigation against any entity by asserting a patent
+     infringement claim (excluding declaratory judgment actions,
+     counter-claims, and cross-claims) alleging that a Contributor Version
+     directly or indirectly infringes any patent, then the rights granted to
+     You by any and all Contributors for the Covered Software under Section
+     2.1 of this License shall terminate.
+
+5.3. In the event of termination under Sections 5.1 or 5.2 above, all end user
+     license agreements (excluding distributors and resellers) which have been
+     validly granted by You or Your distributors under this License prior to
+     termination shall survive termination.
+
+6. Disclaimer of Warranty
+
+   Covered Software is provided under this License on an "as is" basis,
+   without warranty of any kind, either expressed, implied, or statutory,
+   including, without limitation, warranties that the Covered Software is free
+   of defects, merchantable, fit for a particular purpose or non-infringing.
+   The entire risk as to the quality and performance of the Covered Software
+   is with You. Should any Covered Software prove defective in any respect,
+   You (not any Contributor) assume the cost of any necessary servicing,
+   repair, or correction. This disclaimer of warranty constitutes an essential
+   part of this License. No use of  any Covered Software is authorized under
+   this License except under this disclaimer.
+
+7. Limitation of Liability
+
+   Under no circumstances and under no legal theory, whether tort (including
+   negligence), contract, or otherwise, shall any Contributor, or anyone who
+   distributes Covered Software as permitted above, be liable to You for any
+   direct, indirect, special, incidental, or consequential damages of any
+   character including, without limitation, damages for lost profits, loss of
+   goodwill, work stoppage, computer failure or malfunction, or any and all
+   other commercial damages or losses, even if such party shall have been
+   informed of the possibility of such damages. This limitation of liability
+   shall not apply to liability for death or personal injury resulting from
+   such party's negligence to the extent applicable law prohibits such
+   limitation. Some jurisdictions do not allow the exclusion or limitation of
+   incidental or consequential damages, so this exclusion and limitation may
+   not apply to You.
+
+8. Litigation
+
+   Any litigation relating to this License may be brought only in the courts
+   of a jurisdiction where the defendant maintains its principal place of
+   business and such litigation shall be governed by laws of that
+   jurisdiction, without reference to its conflict-of-law provisions. Nothing
+   in this Section shall prevent a party's ability to bring cross-claims or
+   counter-claims.
+
+9. Miscellaneous
+
+   This License represents the complete agreement concerning the subject
+   matter hereof. If any provision of this License is held to be
+   unenforceable, such provision shall be reformed only to the extent
+   necessary to make it enforceable. Any law or regulation which provides that
+   the language of a contract shall be construed against the drafter shall not
+   be used to construe this License against a Contributor.
+
+
+10. Versions of the License
+
+10.1. New Versions
+
+      Mozilla Foundation is the license steward. Except as provided in Section
+      10.3, no one other than the license steward has the right to modify or
+      publish new versions of this License. Each version will be given a
+      distinguishing version number.
+
+10.2. Effect of New Versions
+
+      You may distribute the Covered Software under the terms of the version
+      of the License under which You originally received the Covered Software,
+      or under the terms of any subsequent version published by the license
+      steward.
+
+10.3. Modified Versions
+
+      If you create software not governed by this License, and you want to
+      create a new license for such software, you may create and use a
+      modified version of this License if you rename the license and remove
+      any references to the name of the license steward (except to note that
+      such modified license differs from this License).
+
+10.4. Distributing Source Code Form that is Incompatible With Secondary
+      Licenses If You choose to distribute Source Code Form that is
+      Incompatible With Secondary Licenses under the terms of this version of
+      the License, the notice described in Exhibit B of this License must be
+      attached.
+
+Exhibit A - Source Code Form License Notice
+
+      This Source Code Form is subject to the
+      terms of the Mozilla Public License, v.
+      2.0. If a copy of the MPL was not
+      distributed with this file, You can
+      obtain one at
+      http://mozilla.org/MPL/2.0/.
+
+If it is not possible or desirable to put the notice in a particular file,
+then You may include the notice in a location (such as a LICENSE file in a
+relevant directory) where a recipient would be likely to look for such a
+notice.
+
+You may add additional accurate notices of copyright ownership.
+
+Exhibit B - "Incompatible With Secondary Licenses" Notice
+
+      This Source Code Form is "Incompatible
+      With Secondary Licenses", as defined by
+      the Mozilla Public License, v. 2.0.

vendor/github.com/hashicorp/yamux/addr.go

@@ -0,0 +1,60 @@
+package yamux
+
+import (
+	"fmt"
+	"net"
+)
+
+// hasAddr is used to get the address from the underlying connection
+type hasAddr interface {
+	LocalAddr() net.Addr
+	RemoteAddr() net.Addr
+}
+
+// yamuxAddr is used when we cannot get the underlying address
+type yamuxAddr struct {
+	Addr string
+}
+
+func (*yamuxAddr) Network() string {
+	return "yamux"
+}
+
+func (y *yamuxAddr) String() string {
+	return fmt.Sprintf("yamux:%s", y.Addr)
+}
+
+// Addr is used to get the address of the listener.
+func (s *Session) Addr() net.Addr {
+	return s.LocalAddr()
+}
+
+// LocalAddr is used to get the local address of the
+// underlying connection.
+func (s *Session) LocalAddr() net.Addr {
+	addr, ok := s.conn.(hasAddr)
+	if !ok {
+		return &yamuxAddr{"local"}
+	}
+	return addr.LocalAddr()
+}
+
+// RemoteAddr is used to get the address of remote end
+// of the underlying connection
+func (s *Session) RemoteAddr() net.Addr {
+	addr, ok := s.conn.(hasAddr)
+	if !ok {
+		return &yamuxAddr{"remote"}
+	}
+	return addr.RemoteAddr()
+}
+
+// LocalAddr returns the local address
+func (s *Stream) LocalAddr() net.Addr {
+	return s.session.LocalAddr()
+}
+
+// LocalAddr returns the remote address
+func (s *Stream) RemoteAddr() net.Addr {
+	return s.session.RemoteAddr()
+}

vendor/github.com/hashicorp/yamux/const.go

@@ -0,0 +1,157 @@
+package yamux
+
+import (
+	"encoding/binary"
+	"fmt"
+)
+
+var (
+	// ErrInvalidVersion means we received a frame with an
+	// invalid version
+	ErrInvalidVersion = fmt.Errorf("invalid protocol version")
+
+	// ErrInvalidMsgType means we received a frame with an
+	// invalid message type
+	ErrInvalidMsgType = fmt.Errorf("invalid msg type")
+
+	// ErrSessionShutdown is used if there is a shutdown during
+	// an operation
+	ErrSessionShutdown = fmt.Errorf("session shutdown")
+
+	// ErrStreamsExhausted is returned if we have no more
+	// stream ids to issue
+	ErrStreamsExhausted = fmt.Errorf("streams exhausted")
+
+	// ErrDuplicateStream is used if a duplicate stream is
+	// opened inbound
+	ErrDuplicateStream = fmt.Errorf("duplicate stream initiated")
+
+	// ErrReceiveWindowExceeded indicates the window was exceeded
+	ErrRecvWindowExceeded = fmt.Errorf("recv window exceeded")
+
+	// ErrTimeout is used when we reach an IO deadline
+	ErrTimeout = fmt.Errorf("i/o deadline reached")
+
+	// ErrStreamClosed is returned when using a closed stream
+	ErrStreamClosed = fmt.Errorf("stream closed")
+
+	// ErrUnexpectedFlag is set when we get an unexpected flag
+	ErrUnexpectedFlag = fmt.Errorf("unexpected flag")
+
+	// ErrRemoteGoAway is used when we get a go away from the other side
+	ErrRemoteGoAway = fmt.Errorf("remote end is not accepting connections")
+
+	// ErrConnectionReset is sent if a stream is reset. This can happen
+	// if the backlog is exceeded, or if there was a remote GoAway.
+	ErrConnectionReset = fmt.Errorf("connection reset")
+
+	// ErrConnectionWriteTimeout indicates that we hit the "safety valve"
+	// timeout writing to the underlying stream connection.
+	ErrConnectionWriteTimeout = fmt.Errorf("connection write timeout")
+
+	// ErrKeepAliveTimeout is sent if a missed keepalive caused the stream close
+	ErrKeepAliveTimeout = fmt.Errorf("keepalive timeout")
+)
+
+const (
+	// protoVersion is the only version we support
+	protoVersion uint8 = 0
+)
+
+const (
+	// Data is used for data frames. They are followed
+	// by length bytes worth of payload.
+	typeData uint8 = iota
+
+	// WindowUpdate is used to change the window of
+	// a given stream. The length indicates the delta
+	// update to the window.
+	typeWindowUpdate
+
+	// Ping is sent as a keep-alive or to measure
+	// the RTT. The StreamID and Length value are echoed
+	// back in the response.
+	typePing
+
+	// GoAway is sent to terminate a session. The StreamID
+	// should be 0 and the length is an error code.
+	typeGoAway
+)
+
+const (
+	// SYN is sent to signal a new stream. May
+	// be sent with a data payload
+	flagSYN uint16 = 1 << iota
+
+	// ACK is sent to acknowledge a new stream. May
+	// be sent with a data payload
+	flagACK
+
+	// FIN is sent to half-close the given stream.
+	// May be sent with a data payload.
+	flagFIN
+
+	// RST is used to hard close a given stream.
+	flagRST
+)
+
+const (
+	// initialStreamWindow is the initial stream window size
+	initialStreamWindow uint32 = 256 * 1024
+)
+
+const (
+	// goAwayNormal is sent on a normal termination
+	goAwayNormal uint32 = iota
+
+	// goAwayProtoErr sent on a protocol error
+	goAwayProtoErr
+
+	// goAwayInternalErr sent on an internal error
+	goAwayInternalErr
+)
+
+const (
+	sizeOfVersion  = 1
+	sizeOfType     = 1
+	sizeOfFlags    = 2
+	sizeOfStreamID = 4
+	sizeOfLength   = 4
+	headerSize     = sizeOfVersion + sizeOfType + sizeOfFlags +
+		sizeOfStreamID + sizeOfLength
+)
+
+type header []byte
+
+func (h header) Version() uint8 {
+	return h[0]
+}
+
+func (h header) MsgType() uint8 {
+	return h[1]
+}
+
+func (h header) Flags() uint16 {
+	return binary.BigEndian.Uint16(h[2:4])
+}
+
+func (h header) StreamID() uint32 {
+	return binary.BigEndian.Uint32(h[4:8])
+}
+
+func (h header) Length() uint32 {
+	return binary.BigEndian.Uint32(h[8:12])
+}
+
+func (h header) String() string {
+	return fmt.Sprintf("Vsn:%d Type:%d Flags:%d StreamID:%d Length:%d",
+		h.Version(), h.MsgType(), h.Flags(), h.StreamID(), h.Length())
+}
+
+func (h header) encode(msgType uint8, flags uint16, streamID uint32, length uint32) {
+	h[0] = protoVersion
+	h[1] = msgType
+	binary.BigEndian.PutUint16(h[2:4], flags)
+	binary.BigEndian.PutUint32(h[4:8], streamID)
+	binary.BigEndian.PutUint32(h[8:12], length)
+}

vendor/github.com/hashicorp/yamux/mux.go

@@ -0,0 +1,87 @@
+package yamux
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"time"
+)
+
+// Config is used to tune the Yamux session
+type Config struct {
+	// AcceptBacklog is used to limit how many streams may be
+	// waiting an accept.
+	AcceptBacklog int
+
+	// EnableKeepalive is used to do a period keep alive
+	// messages using a ping.
+	EnableKeepAlive bool
+
+	// KeepAliveInterval is how often to perform the keep alive
+	KeepAliveInterval time.Duration
+
+	// ConnectionWriteTimeout is meant to be a "safety valve" timeout after
+	// we which will suspect a problem with the underlying connection and
+	// close it. This is only applied to writes, where's there's generally
+	// an expectation that things will move along quickly.
+	ConnectionWriteTimeout time.Duration
+
+	// MaxStreamWindowSize is used to control the maximum
+	// window size that we allow for a stream.
+	MaxStreamWindowSize uint32
+
+	// LogOutput is used to control the log destination
+	LogOutput io.Writer
+}
+
+// DefaultConfig is used to return a default configuration
+func DefaultConfig() *Config {
+	return &Config{
+		AcceptBacklog:          256,
+		EnableKeepAlive:        true,
+		KeepAliveInterval:      30 * time.Second,
+		ConnectionWriteTimeout: 10 * time.Second,
+		MaxStreamWindowSize:    initialStreamWindow,
+		LogOutput:              os.Stderr,
+	}
+}
+
+// VerifyConfig is used to verify the sanity of configuration
+func VerifyConfig(config *Config) error {
+	if config.AcceptBacklog <= 0 {
+		return fmt.Errorf("backlog must be positive")
+	}
+	if config.KeepAliveInterval == 0 {
+		return fmt.Errorf("keep-alive interval must be positive")
+	}
+	if config.MaxStreamWindowSize < initialStreamWindow {
+		return fmt.Errorf("MaxStreamWindowSize must be larger than %d", initialStreamWindow)
+	}
+	return nil
+}
+
+// Server is used to initialize a new server-side connection.
+// There must be at most one server-side connection. If a nil config is
+// provided, the DefaultConfiguration will be used.
+func Server(conn io.ReadWriteCloser, config *Config) (*Session, error) {
+	if config == nil {
+		config = DefaultConfig()
+	}
+	if err := VerifyConfig(config); err != nil {
+		return nil, err
+	}
+	return newSession(config, conn, false), nil
+}
+
+// Client is used to initialize a new client-side connection.
+// There must be at most one client-side connection.
+func Client(conn io.ReadWriteCloser, config *Config) (*Session, error) {
+	if config == nil {
+		config = DefaultConfig()
+	}
+
+	if err := VerifyConfig(config); err != nil {
+		return nil, err
+	}
+	return newSession(config, conn, true), nil
+}

vendor/github.com/hashicorp/yamux/session.go

@@ -0,0 +1,623 @@
+package yamux
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"math"
+	"net"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Session is used to wrap a reliable ordered connection and to
+// multiplex it into multiple streams.
+type Session struct {
+	// remoteGoAway indicates the remote side does
+	// not want futher connections. Must be first for alignment.
+	remoteGoAway int32
+
+	// localGoAway indicates that we should stop
+	// accepting futher connections. Must be first for alignment.
+	localGoAway int32
+
+	// nextStreamID is the next stream we should
+	// send. This depends if we are a client/server.
+	nextStreamID uint32
+
+	// config holds our configuration
+	config *Config
+
+	// logger is used for our logs
+	logger *log.Logger
+
+	// conn is the underlying connection
+	conn io.ReadWriteCloser
+
+	// bufRead is a buffered reader
+	bufRead *bufio.Reader
+
+	// pings is used to track inflight pings
+	pings    map[uint32]chan struct{}
+	pingID   uint32
+	pingLock sync.Mutex
+
+	// streams maps a stream id to a stream, and inflight has an entry
+	// for any outgoing stream that has not yet been established. Both are
+	// protected by streamLock.
+	streams    map[uint32]*Stream
+	inflight   map[uint32]struct{}
+	streamLock sync.Mutex
+
+	// synCh acts like a semaphore. It is sized to the AcceptBacklog which
+	// is assumed to be symmetric between the client and server. This allows
+	// the client to avoid exceeding the backlog and instead blocks the open.
+	synCh chan struct{}
+
+	// acceptCh is used to pass ready streams to the client
+	acceptCh chan *Stream
+
+	// sendCh is used to mark a stream as ready to send,
+	// or to send a header out directly.
+	sendCh chan sendReady
+
+	// recvDoneCh is closed when recv() exits to avoid a race
+	// between stream registration and stream shutdown
+	recvDoneCh chan struct{}
+
+	// shutdown is used to safely close a session
+	shutdown     bool
+	shutdownErr  error
+	shutdownCh   chan struct{}
+	shutdownLock sync.Mutex
+}
+
+// sendReady is used to either mark a stream as ready
+// or to directly send a header
+type sendReady struct {
+	Hdr  []byte
+	Body io.Reader
+	Err  chan error
+}
+
+// newSession is used to construct a new session
+func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session {
+	s := &Session{
+		config:     config,
+		logger:     log.New(config.LogOutput, "", log.LstdFlags),
+		conn:       conn,
+		bufRead:    bufio.NewReader(conn),
+		pings:      make(map[uint32]chan struct{}),
+		streams:    make(map[uint32]*Stream),
+		inflight:   make(map[uint32]struct{}),
+		synCh:      make(chan struct{}, config.AcceptBacklog),
+		acceptCh:   make(chan *Stream, config.AcceptBacklog),
+		sendCh:     make(chan sendReady, 64),
+		recvDoneCh: make(chan struct{}),
+		shutdownCh: make(chan struct{}),
+	}
+	if client {
+		s.nextStreamID = 1
+	} else {
+		s.nextStreamID = 2
+	}
+	go s.recv()
+	go s.send()
+	if config.EnableKeepAlive {
+		go s.keepalive()
+	}
+	return s
+}
+
+// IsClosed does a safe check to see if we have shutdown
+func (s *Session) IsClosed() bool {
+	select {
+	case <-s.shutdownCh:
+		return true
+	default:
+		return false
+	}
+}
+
+// NumStreams returns the number of currently open streams
+func (s *Session) NumStreams() int {
+	s.streamLock.Lock()
+	num := len(s.streams)
+	s.streamLock.Unlock()
+	return num
+}
+
+// Open is used to create a new stream as a net.Conn
+func (s *Session) Open() (net.Conn, error) {
+	conn, err := s.OpenStream()
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+
+// OpenStream is used to create a new stream
+func (s *Session) OpenStream() (*Stream, error) {
+	if s.IsClosed() {
+		return nil, ErrSessionShutdown
+	}
+	if atomic.LoadInt32(&s.remoteGoAway) == 1 {
+		return nil, ErrRemoteGoAway
+	}
+
+	// Block if we have too many inflight SYNs
+	select {
+	case s.synCh <- struct{}{}:
+	case <-s.shutdownCh:
+		return nil, ErrSessionShutdown
+	}
+
+GET_ID:
+	// Get an ID, and check for stream exhaustion
+	id := atomic.LoadUint32(&s.nextStreamID)
+	if id >= math.MaxUint32-1 {
+		return nil, ErrStreamsExhausted
+	}
+	if !atomic.CompareAndSwapUint32(&s.nextStreamID, id, id+2) {
+		goto GET_ID
+	}
+
+	// Register the stream
+	stream := newStream(s, id, streamInit)
+	s.streamLock.Lock()
+	s.streams[id] = stream
+	s.inflight[id] = struct{}{}
+	s.streamLock.Unlock()
+
+	// Send the window update to create
+	if err := stream.sendWindowUpdate(); err != nil {
+		select {
+		case <-s.synCh:
+		default:
+			s.logger.Printf("[ERR] yamux: aborted stream open without inflight syn semaphore")
+		}
+		return nil, err
+	}
+	return stream, nil
+}
+
+// Accept is used to block until the next available stream
+// is ready to be accepted.
+func (s *Session) Accept() (net.Conn, error) {
+	conn, err := s.AcceptStream()
+	if err != nil {
+		return nil, err
+	}
+	return conn, err
+}
+
+// AcceptStream is used to block until the next available stream
+// is ready to be accepted.
+func (s *Session) AcceptStream() (*Stream, error) {
+	select {
+	case stream := <-s.acceptCh:
+		if err := stream.sendWindowUpdate(); err != nil {
+			return nil, err
+		}
+		return stream, nil
+	case <-s.shutdownCh:
+		return nil, s.shutdownErr
+	}
+}
+
+// Close is used to close the session and all streams.
+// Attempts to send a GoAway before closing the connection.
+func (s *Session) Close() error {
+	s.shutdownLock.Lock()
+	defer s.shutdownLock.Unlock()
+
+	if s.shutdown {
+		return nil
+	}
+	s.shutdown = true
+	if s.shutdownErr == nil {
+		s.shutdownErr = ErrSessionShutdown
+	}
+	close(s.shutdownCh)
+	s.conn.Close()
+	<-s.recvDoneCh
+
+	s.streamLock.Lock()
+	defer s.streamLock.Unlock()
+	for _, stream := range s.streams {
+		stream.forceClose()
+	}
+	return nil
+}
+
+// exitErr is used to handle an error that is causing the
+// session to terminate.
+func (s *Session) exitErr(err error) {
+	s.shutdownLock.Lock()
+	if s.shutdownErr == nil {
+		s.shutdownErr = err
+	}
+	s.shutdownLock.Unlock()
+	s.Close()
+}
+
+// GoAway can be used to prevent accepting further
+// connections. It does not close the underlying conn.
+func (s *Session) GoAway() error {
+	return s.waitForSend(s.goAway(goAwayNormal), nil)
+}
+
+// goAway is used to send a goAway message
+func (s *Session) goAway(reason uint32) header {
+	atomic.SwapInt32(&s.localGoAway, 1)
+	hdr := header(make([]byte, headerSize))
+	hdr.encode(typeGoAway, 0, 0, reason)
+	return hdr
+}
+
+// Ping is used to measure the RTT response time
+func (s *Session) Ping() (time.Duration, error) {
+	// Get a channel for the ping
+	ch := make(chan struct{})
+
+	// Get a new ping id, mark as pending
+	s.pingLock.Lock()
+	id := s.pingID
+	s.pingID++
+	s.pings[id] = ch
+	s.pingLock.Unlock()
+
+	// Send the ping request
+	hdr := header(make([]byte, headerSize))
+	hdr.encode(typePing, flagSYN, 0, id)
+	if err := s.waitForSend(hdr, nil); err != nil {
+		return 0, err
+	}
+
+	// Wait for a response
+	start := time.Now()
+	select {
+	case <-ch:
+	case <-time.After(s.config.ConnectionWriteTimeout):
+		s.pingLock.Lock()
+		delete(s.pings, id) // Ignore it if a response comes later.
+		s.pingLock.Unlock()
+		return 0, ErrTimeout
+	case <-s.shutdownCh:
+		return 0, ErrSessionShutdown
+	}
+
+	// Compute the RTT
+	return time.Now().Sub(start), nil
+}
+
+// keepalive is a long running goroutine that periodically does
+// a ping to keep the connection alive.
+func (s *Session) keepalive() {
+	for {
+		select {
+		case <-time.After(s.config.KeepAliveInterval):
+			_, err := s.Ping()
+			if err != nil {
+				s.logger.Printf("[ERR] yamux: keepalive failed: %v", err)
+				s.exitErr(ErrKeepAliveTimeout)
+				return
+			}
+		case <-s.shutdownCh:
+			return
+		}
+	}
+}
+
+// waitForSendErr waits to send a header, checking for a potential shutdown
+func (s *Session) waitForSend(hdr header, body io.Reader) error {
+	errCh := make(chan error, 1)
+	return s.waitForSendErr(hdr, body, errCh)
+}
+
+// waitForSendErr waits to send a header with optional data, checking for a
+// potential shutdown. Since there's the expectation that sends can happen
+// in a timely manner, we enforce the connection write timeout here.
+func (s *Session) waitForSendErr(hdr header, body io.Reader, errCh chan error) error {
+	timer := time.NewTimer(s.config.ConnectionWriteTimeout)
+	defer timer.Stop()
+
+	ready := sendReady{Hdr: hdr, Body: body, Err: errCh}
+	select {
+	case s.sendCh <- ready:
+	case <-s.shutdownCh:
+		return ErrSessionShutdown
+	case <-timer.C:
+		return ErrConnectionWriteTimeout
+	}
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-s.shutdownCh:
+		return ErrSessionShutdown
+	case <-timer.C:
+		return ErrConnectionWriteTimeout
+	}
+}
+
+// sendNoWait does a send without waiting. Since there's the expectation that
+// the send happens right here, we enforce the connection write timeout if we
+// can't queue the header to be sent.
+func (s *Session) sendNoWait(hdr header) error {
+	timer := time.NewTimer(s.config.ConnectionWriteTimeout)
+	defer timer.Stop()
+
+	select {
+	case s.sendCh <- sendReady{Hdr: hdr}:
+		return nil
+	case <-s.shutdownCh:
+		return ErrSessionShutdown
+	case <-timer.C:
+		return ErrConnectionWriteTimeout
+	}
+}
+
+// send is a long running goroutine that sends data
+func (s *Session) send() {
+	for {
+		select {
+		case ready := <-s.sendCh:
+			// Send a header if ready
+			if ready.Hdr != nil {
+				sent := 0
+				for sent < len(ready.Hdr) {
+					n, err := s.conn.Write(ready.Hdr[sent:])
+					if err != nil {
+						s.logger.Printf("[ERR] yamux: Failed to write header: %v", err)
+						asyncSendErr(ready.Err, err)
+						s.exitErr(err)
+						return
+					}
+					sent += n
+				}
+			}
+
+			// Send data from a body if given
+			if ready.Body != nil {
+				_, err := io.Copy(s.conn, ready.Body)
+				if err != nil {
+					s.logger.Printf("[ERR] yamux: Failed to write body: %v", err)
+					asyncSendErr(ready.Err, err)
+					s.exitErr(err)
+					return
+				}
+			}
+
+			// No error, successful send
+			asyncSendErr(ready.Err, nil)
+		case <-s.shutdownCh:
+			return
+		}
+	}
+}
+
+// recv is a long running goroutine that accepts new data
+func (s *Session) recv() {
+	if err := s.recvLoop(); err != nil {
+		s.exitErr(err)
+	}
+}
+
+// recvLoop continues to receive data until a fatal error is encountered
+func (s *Session) recvLoop() error {
+	defer close(s.recvDoneCh)
+	hdr := header(make([]byte, headerSize))
+	var handler func(header) error
+	for {
+		// Read the header
+		if _, err := io.ReadFull(s.bufRead, hdr); err != nil {
+			if err != io.EOF && !strings.Contains(err.Error(), "closed") && !strings.Contains(err.Error(), "reset by peer") {
+				s.logger.Printf("[ERR] yamux: Failed to read header: %v", err)
+			}
+			return err
+		}
+
+		// Verify the version
+		if hdr.Version() != protoVersion {
+			s.logger.Printf("[ERR] yamux: Invalid protocol version: %d", hdr.Version())
+			return ErrInvalidVersion
+		}
+
+		// Switch on the type
+		switch hdr.MsgType() {
+		case typeData:
+			handler = s.handleStreamMessage
+		case typeWindowUpdate:
+			handler = s.handleStreamMessage
+		case typeGoAway:
+			handler = s.handleGoAway
+		case typePing:
+			handler = s.handlePing
+		default:
+			return ErrInvalidMsgType
+		}
+
+		// Invoke the handler
+		if err := handler(hdr); err != nil {
+			return err
+		}
+	}
+}
+
+// handleStreamMessage handles either a data or window update frame
+func (s *Session) handleStreamMessage(hdr header) error {
+	// Check for a new stream creation
+	id := hdr.StreamID()
+	flags := hdr.Flags()
+	if flags&flagSYN == flagSYN {
+		if err := s.incomingStream(id); err != nil {
+			return err
+		}
+	}
+
+	// Get the stream
+	s.streamLock.Lock()
+	stream := s.streams[id]
+	s.streamLock.Unlock()
+
+	// If we do not have a stream, likely we sent a RST
+	if stream == nil {
+		// Drain any data on the wire
+		if hdr.MsgType() == typeData && hdr.Length() > 0 {
+			s.logger.Printf("[WARN] yamux: Discarding data for stream: %d", id)
+			if _, err := io.CopyN(ioutil.Discard, s.bufRead, int64(hdr.Length())); err != nil {
+				s.logger.Printf("[ERR] yamux: Failed to discard data: %v", err)
+				return nil
+			}
+		} else {
+			s.logger.Printf("[WARN] yamux: frame for missing stream: %v", hdr)
+		}
+		return nil
+	}
+
+	// Check if this is a window update
+	if hdr.MsgType() == typeWindowUpdate {
+		if err := stream.incrSendWindow(hdr, flags); err != nil {
+			if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
+				s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
+			}
+			return err
+		}
+		return nil
+	}
+
+	// Read the new data
+	if err := stream.readData(hdr, flags, s.bufRead); err != nil {
+		if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
+			s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
+		}
+		return err
+	}
+	return nil
+}
+
+// handlePing is invokde for a typePing frame
+func (s *Session) handlePing(hdr header) error {
+	flags := hdr.Flags()
+	pingID := hdr.Length()
+
+	// Check if this is a query, respond back in a separate context so we
+	// don't interfere with the receiving thread blocking for the write.
+	if flags&flagSYN == flagSYN {
+		go func() {
+			hdr := header(make([]byte, headerSize))
+			hdr.encode(typePing, flagACK, 0, pingID)
+			if err := s.sendNoWait(hdr); err != nil {
+				s.logger.Printf("[WARN] yamux: failed to send ping reply: %v", err)
+			}
+		}()
+		return nil
+	}
+
+	// Handle a response
+	s.pingLock.Lock()
+	ch := s.pings[pingID]
+	if ch != nil {
+		delete(s.pings, pingID)
+		close(ch)
+	}
+	s.pingLock.Unlock()
+	return nil
+}
+
+// handleGoAway is invokde for a typeGoAway frame
+func (s *Session) handleGoAway(hdr header) error {
+	code := hdr.Length()
+	switch code {
+	case goAwayNormal:
+		atomic.SwapInt32(&s.remoteGoAway, 1)
+	case goAwayProtoErr:
+		s.logger.Printf("[ERR] yamux: received protocol error go away")
+		return fmt.Errorf("yamux protocol error")
+	case goAwayInternalErr:
+		s.logger.Printf("[ERR] yamux: received internal error go away")
+		return fmt.Errorf("remote yamux internal error")
+	default:
+		s.logger.Printf("[ERR] yamux: received unexpected go away")
+		return fmt.Errorf("unexpected go away received")
+	}
+	return nil
+}
+
+// incomingStream is used to create a new incoming stream
+func (s *Session) incomingStream(id uint32) error {
+	// Reject immediately if we are doing a go away
+	if atomic.LoadInt32(&s.localGoAway) == 1 {
+		hdr := header(make([]byte, headerSize))
+		hdr.encode(typeWindowUpdate, flagRST, id, 0)
+		return s.sendNoWait(hdr)
+	}
+
+	// Allocate a new stream
+	stream := newStream(s, id, streamSYNReceived)
+
+	s.streamLock.Lock()
+	defer s.streamLock.Unlock()
+
+	// Check if stream already exists
+	if _, ok := s.streams[id]; ok {
+		s.logger.Printf("[ERR] yamux: duplicate stream declared")
+		if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
+			s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
+		}
+		return ErrDuplicateStream
+	}
+
+	// Register the stream
+	s.streams[id] = stream
+
+	// Check if we've exceeded the backlog
+	select {
+	case s.acceptCh <- stream:
+		return nil
+	default:
+		// Backlog exceeded! RST the stream
+		s.logger.Printf("[WARN] yamux: backlog exceeded, forcing connection reset")
+		delete(s.streams, id)
+		stream.sendHdr.encode(typeWindowUpdate, flagRST, id, 0)
+		return s.sendNoWait(stream.sendHdr)
+	}
+}
+
+// closeStream is used to close a stream once both sides have
+// issued a close. If there was an in-flight SYN and the stream
+// was not yet established, then this will give the credit back.
+func (s *Session) closeStream(id uint32) {
+	s.streamLock.Lock()
+	if _, ok := s.inflight[id]; ok {
+		select {
+		case <-s.synCh:
+		default:
+			s.logger.Printf("[ERR] yamux: SYN tracking out of sync")
+		}
+	}
+	delete(s.streams, id)
+	s.streamLock.Unlock()
+}
+
+// establishStream is used to mark a stream that was in the
+// SYN Sent state as established.
+func (s *Session) establishStream(id uint32) {
+	s.streamLock.Lock()
+	if _, ok := s.inflight[id]; ok {
+		delete(s.inflight, id)
+	} else {
+		s.logger.Printf("[ERR] yamux: established stream without inflight SYN (no tracking entry)")
+	}
+	select {
+	case <-s.synCh:
+	default:
+		s.logger.Printf("[ERR] yamux: established stream without inflight SYN (didn't have semaphore)")
+	}
+	s.streamLock.Unlock()
+}

vendor/github.com/hashicorp/yamux/stream.go

@@ -0,0 +1,457 @@
+package yamux
+
+import (
+	"bytes"
+	"io"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type streamState int
+
+const (
+	streamInit streamState = iota
+	streamSYNSent
+	streamSYNReceived
+	streamEstablished
+	streamLocalClose
+	streamRemoteClose
+	streamClosed
+	streamReset
+)
+
+// Stream is used to represent a logical stream
+// within a session.
+type Stream struct {
+	recvWindow uint32
+	sendWindow uint32
+
+	id      uint32
+	session *Session
+
+	state     streamState
+	stateLock sync.Mutex
+
+	recvBuf  *bytes.Buffer
+	recvLock sync.Mutex
+
+	controlHdr     header
+	controlErr     chan error
+	controlHdrLock sync.Mutex
+
+	sendHdr  header
+	sendErr  chan error
+	sendLock sync.Mutex
+
+	recvNotifyCh chan struct{}
+	sendNotifyCh chan struct{}
+
+	readDeadline  time.Time
+	writeDeadline time.Time
+}
+
+// newStream is used to construct a new stream within
+// a given session for an ID
+func newStream(session *Session, id uint32, state streamState) *Stream {
+	s := &Stream{
+		id:           id,
+		session:      session,
+		state:        state,
+		controlHdr:   header(make([]byte, headerSize)),
+		controlErr:   make(chan error, 1),
+		sendHdr:      header(make([]byte, headerSize)),
+		sendErr:      make(chan error, 1),
+		recvWindow:   initialStreamWindow,
+		sendWindow:   initialStreamWindow,
+		recvNotifyCh: make(chan struct{}, 1),
+		sendNotifyCh: make(chan struct{}, 1),
+	}
+	return s
+}
+
+// Session returns the associated stream session
+func (s *Stream) Session() *Session {
+	return s.session
+}
+
+// StreamID returns the ID of this stream
+func (s *Stream) StreamID() uint32 {
+	return s.id
+}
+
+// Read is used to read from the stream
+func (s *Stream) Read(b []byte) (n int, err error) {
+	defer asyncNotify(s.recvNotifyCh)
+START:
+	s.stateLock.Lock()
+	switch s.state {
+	case streamLocalClose:
+		fallthrough
+	case streamRemoteClose:
+		fallthrough
+	case streamClosed:
+		s.recvLock.Lock()
+		if s.recvBuf == nil || s.recvBuf.Len() == 0 {
+			s.recvLock.Unlock()
+			s.stateLock.Unlock()
+			return 0, io.EOF
+		}
+		s.recvLock.Unlock()
+	case streamReset:
+		s.stateLock.Unlock()
+		return 0, ErrConnectionReset
+	}
+	s.stateLock.Unlock()
+
+	// If there is no data available, block
+	s.recvLock.Lock()
+	if s.recvBuf == nil || s.recvBuf.Len() == 0 {
+		s.recvLock.Unlock()
+		goto WAIT
+	}
+
+	// Read any bytes
+	n, _ = s.recvBuf.Read(b)
+	s.recvLock.Unlock()
+
+	// Send a window update potentially
+	err = s.sendWindowUpdate()
+	return n, err
+
+WAIT:
+	var timeout <-chan time.Time
+	var timer *time.Timer
+	if !s.readDeadline.IsZero() {
+		delay := s.readDeadline.Sub(time.Now())
+		timer = time.NewTimer(delay)
+		timeout = timer.C
+	}
+	select {
+	case <-s.recvNotifyCh:
+		if timer != nil {
+			timer.Stop()
+		}
+		goto START
+	case <-timeout:
+		return 0, ErrTimeout
+	}
+}
+
+// Write is used to write to the stream
+func (s *Stream) Write(b []byte) (n int, err error) {
+	s.sendLock.Lock()
+	defer s.sendLock.Unlock()
+	total := 0
+	for total < len(b) {
+		n, err := s.write(b[total:])
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+	return total, nil
+}
+
+// write is used to write to the stream, may return on
+// a short write.
+func (s *Stream) write(b []byte) (n int, err error) {
+	var flags uint16
+	var max uint32
+	var body io.Reader
+START:
+	s.stateLock.Lock()
+	switch s.state {
+	case streamLocalClose:
+		fallthrough
+	case streamClosed:
+		s.stateLock.Unlock()
+		return 0, ErrStreamClosed
+	case streamReset:
+		s.stateLock.Unlock()
+		return 0, ErrConnectionReset
+	}
+	s.stateLock.Unlock()
+
+	// If there is no data available, block
+	window := atomic.LoadUint32(&s.sendWindow)
+	if window == 0 {
+		goto WAIT
+	}
+
+	// Determine the flags if any
+	flags = s.sendFlags()
+
+	// Send up to our send window
+	max = min(window, uint32(len(b)))
+	body = bytes.NewReader(b[:max])
+
+	// Send the header
+	s.sendHdr.encode(typeData, flags, s.id, max)
+	if err := s.session.waitForSendErr(s.sendHdr, body, s.sendErr); err != nil {
+		return 0, err
+	}
+
+	// Reduce our send window
+	atomic.AddUint32(&s.sendWindow, ^uint32(max-1))
+
+	// Unlock
+	return int(max), err
+
+WAIT:
+	var timeout <-chan time.Time
+	if !s.writeDeadline.IsZero() {
+		delay := s.writeDeadline.Sub(time.Now())
+		timeout = time.After(delay)
+	}
+	select {
+	case <-s.sendNotifyCh:
+		goto START
+	case <-timeout:
+		return 0, ErrTimeout
+	}
+	return 0, nil
+}
+
+// sendFlags determines any flags that are appropriate
+// based on the current stream state
+func (s *Stream) sendFlags() uint16 {
+	s.stateLock.Lock()
+	defer s.stateLock.Unlock()
+	var flags uint16
+	switch s.state {
+	case streamInit:
+		flags |= flagSYN
+		s.state = streamSYNSent
+	case streamSYNReceived:
+		flags |= flagACK
+		s.state = streamEstablished
+	}
+	return flags
+}
+
+// sendWindowUpdate potentially sends a window update enabling
+// further writes to take place. Must be invoked with the lock.
+func (s *Stream) sendWindowUpdate() error {
+	s.controlHdrLock.Lock()
+	defer s.controlHdrLock.Unlock()
+
+	// Determine the delta update
+	max := s.session.config.MaxStreamWindowSize
+	delta := max - atomic.LoadUint32(&s.recvWindow)
+
+	// Determine the flags if any
+	flags := s.sendFlags()
+
+	// Check if we can omit the update
+	if delta < (max/2) && flags == 0 {
+		return nil
+	}
+
+	// Update our window
+	atomic.AddUint32(&s.recvWindow, delta)
+
+	// Send the header
+	s.controlHdr.encode(typeWindowUpdate, flags, s.id, delta)
+	if err := s.session.waitForSendErr(s.controlHdr, nil, s.controlErr); err != nil {
+		return err
+	}
+	return nil
+}
+
+// sendClose is used to send a FIN
+func (s *Stream) sendClose() error {
+	s.controlHdrLock.Lock()
+	defer s.controlHdrLock.Unlock()
+
+	flags := s.sendFlags()
+	flags |= flagFIN
+	s.controlHdr.encode(typeWindowUpdate, flags, s.id, 0)
+	if err := s.session.waitForSendErr(s.controlHdr, nil, s.controlErr); err != nil {
+		return err
+	}
+	return nil
+}
+
+// Close is used to close the stream
+func (s *Stream) Close() error {
+	closeStream := false
+	s.stateLock.Lock()
+	switch s.state {
+	// Opened means we need to signal a close
+	case streamSYNSent:
+		fallthrough
+	case streamSYNReceived:
+		fallthrough
+	case streamEstablished:
+		s.state = streamLocalClose
+		goto SEND_CLOSE
+
+	case streamLocalClose:
+	case streamRemoteClose:
+		s.state = streamClosed
+		closeStream = true
+		goto SEND_CLOSE
+
+	case streamClosed:
+	case streamReset:
+	default:
+		panic("unhandled state")
+	}
+	s.stateLock.Unlock()
+	return nil
+SEND_CLOSE:
+	s.stateLock.Unlock()
+	s.sendClose()
+	s.notifyWaiting()
+	if closeStream {
+		s.session.closeStream(s.id)
+	}
+	return nil
+}
+
+// forceClose is used for when the session is exiting
+func (s *Stream) forceClose() {
+	s.stateLock.Lock()
+	s.state = streamClosed
+	s.stateLock.Unlock()
+	s.notifyWaiting()
+}
+
+// processFlags is used to update the state of the stream
+// based on set flags, if any. Lock must be held
+func (s *Stream) processFlags(flags uint16) error {
+	// Close the stream without holding the state lock
+	closeStream := false
+	defer func() {
+		if closeStream {
+			s.session.closeStream(s.id)
+		}
+	}()
+
+	s.stateLock.Lock()
+	defer s.stateLock.Unlock()
+	if flags&flagACK == flagACK {
+		if s.state == streamSYNSent {
+			s.state = streamEstablished
+		}
+		s.session.establishStream(s.id)
+	}
+	if flags&flagFIN == flagFIN {
+		switch s.state {
+		case streamSYNSent:
+			fallthrough
+		case streamSYNReceived:
+			fallthrough
+		case streamEstablished:
+			s.state = streamRemoteClose
+			s.notifyWaiting()
+		case streamLocalClose:
+			s.state = streamClosed
+			closeStream = true
+			s.notifyWaiting()
+		default:
+			s.session.logger.Printf("[ERR] yamux: unexpected FIN flag in state %d", s.state)
+			return ErrUnexpectedFlag
+		}
+	}
+	if flags&flagRST == flagRST {
+		s.state = streamReset
+		closeStream = true
+		s.notifyWaiting()
+	}
+	return nil
+}
+
+// notifyWaiting notifies all the waiting channels
+func (s *Stream) notifyWaiting() {
+	asyncNotify(s.recvNotifyCh)
+	asyncNotify(s.sendNotifyCh)
+}
+
+// incrSendWindow updates the size of our send window
+func (s *Stream) incrSendWindow(hdr header, flags uint16) error {
+	if err := s.processFlags(flags); err != nil {
+		return err
+	}
+
+	// Increase window, unblock a sender
+	atomic.AddUint32(&s.sendWindow, hdr.Length())
+	asyncNotify(s.sendNotifyCh)
+	return nil
+}
+
+// readData is used to handle a data frame
+func (s *Stream) readData(hdr header, flags uint16, conn io.Reader) error {
+	if err := s.processFlags(flags); err != nil {
+		return err
+	}
+
+	// Check that our recv window is not exceeded
+	length := hdr.Length()
+	if length == 0 {
+		return nil
+	}
+	if remain := atomic.LoadUint32(&s.recvWindow); length > remain {
+		s.session.logger.Printf("[ERR] yamux: receive window exceeded (stream: %d, remain: %d, recv: %d)", s.id, remain, length)
+		return ErrRecvWindowExceeded
+	}
+
+	// Wrap in a limited reader
+	conn = &io.LimitedReader{R: conn, N: int64(length)}
+
+	// Copy into buffer
+	s.recvLock.Lock()
+	if s.recvBuf == nil {
+		// Allocate the receive buffer just-in-time to fit the full data frame.
+		// This way we can read in the whole packet without further allocations.
+		s.recvBuf = bytes.NewBuffer(make([]byte, 0, length))
+	}
+	if _, err := io.Copy(s.recvBuf, conn); err != nil {
+		s.session.logger.Printf("[ERR] yamux: Failed to read stream data: %v", err)
+		s.recvLock.Unlock()
+		return err
+	}
+
+	// Decrement the receive window
+	atomic.AddUint32(&s.recvWindow, ^uint32(length-1))
+	s.recvLock.Unlock()
+
+	// Unblock any readers
+	asyncNotify(s.recvNotifyCh)
+	return nil
+}
+
+// SetDeadline sets the read and write deadlines
+func (s *Stream) SetDeadline(t time.Time) error {
+	if err := s.SetReadDeadline(t); err != nil {
+		return err
+	}
+	if err := s.SetWriteDeadline(t); err != nil {
+		return err
+	}
+	return nil
+}
+
+// SetReadDeadline sets the deadline for future Read calls.
+func (s *Stream) SetReadDeadline(t time.Time) error {
+	s.readDeadline = t
+	return nil
+}
+
+// SetWriteDeadline sets the deadline for future Write calls
+func (s *Stream) SetWriteDeadline(t time.Time) error {
+	s.writeDeadline = t
+	return nil
+}
+
+// Shrink is used to compact the amount of buffers utilized
+// This is useful when using Yamux in a connection pool to reduce
+// the idle memory utilization.
+func (s *Stream) Shrink() {
+	s.recvLock.Lock()
+	if s.recvBuf != nil && s.recvBuf.Len() == 0 {
+		s.recvBuf = nil
+	}
+	s.recvLock.Unlock()
+}

vendor/github.com/hashicorp/yamux/util.go

@@ -0,0 +1,28 @@
+package yamux
+
+// asyncSendErr is used to try an async send of an error
+func asyncSendErr(ch chan error, err error) {
+	if ch == nil {
+		return
+	}
+	select {
+	case ch <- err:
+	default:
+	}
+}
+
+// asyncNotify is used to signal a waiting goroutine
+func asyncNotify(ch chan struct{}) {
+	select {
+	case ch <- struct{}{}:
+	default:
+	}
+}
+
+// min computes the minimum of two values
+func min(a, b uint32) uint32 {
+	if a < b {
+		return a
+	}
+	return b
+}

vendor/github.com/kata-containers/agent/protocols/client/client.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/hashicorp/yamux"
 	"github.com/mdlayher/vsock"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -25,9 +26,10 @@ import (
 const (
 	unixSocketScheme  = "unix"
 	vsockSocketScheme = "vsock"
-	dialTimeout       = 5 * time.Second
 )
 
+var defaultDialTimeout = 5 * time.Second
+
 // AgentClient is an agent gRPC client connection wrapper for agentgrpc.AgentServiceClient
 type AgentClient struct {
 	agentgrpc.AgentServiceClient
@@ -43,15 +45,15 @@ type dialer func(string, time.Duration) (net.Conn, error)
 //   - unix://<unix socket path>
 //   - vsock://<cid>:<port>
 //   - <unix socket path>
-func NewAgentClient(sock string) (*AgentClient, error) {
+func NewAgentClient(sock string, enableYamux bool) (*AgentClient, error) {
 	grpcAddr, parsedAddr, err := parse(sock)
 	if err != nil {
 		return nil, err
 	}
 	dialOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock()}
-	dialOpts = append(dialOpts, grpc.WithDialer(agentDialer(parsedAddr)))
+	dialOpts = append(dialOpts, grpc.WithDialer(agentDialer(parsedAddr, enableYamux)))
 	ctx := context.Background()
-	ctx, cancel := context.WithTimeout(ctx, dialTimeout)
+	ctx, cancel := context.WithTimeout(ctx, defaultDialTimeout)
 	defer cancel()
 	conn, err := grpc.DialContext(ctx, grpcAddr, dialOpts...)
 	if err != nil {
@@ -117,14 +119,46 @@ func parse(sock string) (string, *url.URL, error) {
 	return grpcAddr, addr, nil
 }
 
-func agentDialer(addr *url.URL) dialer {
+func agentDialer(addr *url.URL, enableYamux bool) dialer {
+	var d dialer
 	switch addr.Scheme {
 	case vsockSocketScheme:
-		return vsockDialer
+		d = vsockDialer
 	case unixSocketScheme:
 		fallthrough
 	default:
-		return unixDialer
+		d = unixDialer
+	}
+
+	if !enableYamux {
+		return d
+	}
+
+	// yamux dialer
+	return func(sock string, timeout time.Duration) (net.Conn, error) {
+		conn, err := d(sock, timeout)
+		if err != nil {
+			return nil, err
+		}
+		defer func() {
+			if err != nil {
+				conn.Close()
+			}
+		}()
+
+		var session *yamux.Session
+		session, err = yamux.Client(conn, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		var stream net.Conn
+		stream, err = session.Open()
+		if err != nil {
+			return nil, err
+		}
+
+		return stream, nil
 	}
 }
 

virtcontainers/api_test.go

@@ -1029,6 +1029,7 @@ func TestStatusPodFailingFetchPodConfig(t *testing.T) {
 
 	path := filepath.Join(configStoragePath, p.ID())
 	os.RemoveAll(path)
+	globalPodList.removePod(p.ID())
 
 	_, err = StatusPod(p.ID())
 	if err == nil {
@@ -1050,6 +1051,7 @@ func TestStatusPodPodFailingFetchPodState(t *testing.T) {
 	assert.True(t, ok)
 
 	os.RemoveAll(pImpl.configPath)
+	globalPodList.removePod(p.ID())
 
 	_, err = StatusPod(p.ID())
 	if err == nil {
@@ -1932,6 +1934,7 @@ func TestStatusContainerFailing(t *testing.T) {
 	assert.True(t, ok)
 
 	os.RemoveAll(pImpl.configPath)
+	globalPodList.removePod(p.ID())
 
 	_, err = StatusContainer(p.ID(), contID)
 	if err == nil {

virtcontainers/kata_agent.go

@@ -73,8 +73,9 @@ func (s *kataVSOCK) String() string {
 // KataAgentState is the structure describing the data stored from this
 // agent implementation.
 type KataAgentState struct {
-	ProxyPid int
+	ProxyPid     int
-	URL      string
+	ProxyBuiltIn bool
+	URL          string
 }
 
 type kataAgent struct {
@@ -417,6 +418,7 @@ func (k *kataAgent) startPod(pod Pod) error {
 
 	proxyParams := proxyParams{
 		agentURL: agentURL,
+		logger:   k.Logger().WithField("pod-id", pod.id),
 	}
 
 	// Start the proxy here
@@ -427,12 +429,17 @@ func (k *kataAgent) startPod(pod Pod) error {
 
 	// Fill agent state with proxy information, and store them.
 	k.state.ProxyPid = pid
+	k.state.ProxyBuiltIn = isProxyBuiltIn(pod.config.ProxyType)
 	k.state.URL = uri
 	if err := pod.storage.storeAgentState(pod.id, k.state); err != nil {
 		return err
 	}
 
-	k.Logger().WithField("proxy-pid", pid).Info("proxy started")
+	k.Logger().WithFields(logrus.Fields{
+		"pod-id":    pod.id,
+		"proxy-pid": pid,
+		"proxy-url": uri,
+	}).Info("proxy started")
 
 	hostname := pod.config.Hostname
 	if len(hostname) > maxHostnameLen {
@@ -805,7 +812,7 @@ func (k *kataAgent) connect() error {
 		return nil
 	}
 
-	client, err := kataclient.NewAgentClient(k.state.URL)
+	client, err := kataclient.NewAgentClient(k.state.URL, k.state.ProxyBuiltIn)
 	if err != nil {
 		return err
 	}

virtcontainers/kata_builtin_proxy.go

@@ -0,0 +1,103 @@
+//
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// 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 virtcontainers
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+
+	"github.com/sirupsen/logrus"
+)
+
+// This is a kata builtin proxy implementation of the proxy interface. Kata proxy
+// functionality is implemented inside the virtcontainers library.
+type kataBuiltInProxy struct {
+	podID string
+	conn  net.Conn
+}
+
+// start is the proxy start implementation for kata builtin proxy.
+// It starts the console watcher for the guest.
+// It returns agentURL to let agent connect directly.
+func (p *kataBuiltInProxy) start(pod Pod, params proxyParams) (int, string, error) {
+	if p.conn != nil {
+		return -1, "", fmt.Errorf("kata builtin proxy running for pod %s", p.podID)
+	}
+
+	p.podID = pod.id
+	console := pod.hypervisor.getPodConsole(pod.id)
+	err := p.watchConsole(consoleProtoUnix, console, params.logger)
+	if err != nil {
+		return -1, "", err
+	}
+
+	return -1, params.agentURL, nil
+}
+
+// stop is the proxy stop implementation for kata builtin proxy.
+func (p *kataBuiltInProxy) stop(pod Pod, pid int) error {
+	if p.conn != nil {
+		p.conn.Close()
+		p.conn = nil
+		p.podID = ""
+	}
+	return nil
+}
+
+func (p *kataBuiltInProxy) watchConsole(proto, console string, logger *logrus.Entry) (err error) {
+	var (
+		scanner *bufio.Scanner
+		conn    net.Conn
+	)
+
+	switch proto {
+	case consoleProtoUnix:
+		conn, err = net.Dial("unix", console)
+		if err != nil {
+			return err
+		}
+	// TODO: add pty console support for kvmtools
+	case consoleProtoPty:
+		fallthrough
+	default:
+		return fmt.Errorf("unknown console proto %s", proto)
+	}
+
+	p.conn = conn
+
+	go func() {
+		scanner = bufio.NewScanner(conn)
+		for scanner.Scan() {
+			fmt.Printf("[POD-%s] vmconsole: %s\n", p.podID, scanner.Text())
+		}
+
+		if err := scanner.Err(); err != nil {
+			if err == io.EOF {
+				logger.Info("console watcher quits")
+			} else {
+				logger.WithError(err).WithFields(logrus.Fields{
+					"console-protocol": proto,
+					"console-socket":   console,
+				}).Error("Failed to read agent logs")
+			}
+		}
+	}()
+
+	return nil
+}

virtcontainers/kata_builtin_shim.go

@@ -0,0 +1,26 @@
+//
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// 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 virtcontainers
+
+type kataBuiltInShim struct{}
+
+// start is the kataBuiltInShim start implementation for kata builtin shim.
+// It does nothing. The shim functionality is provided by the virtcontainers
+// library.
+func (s *kataBuiltInShim) start(pod Pod, params ShimParams) (int, error) {
+	return -1, nil
+}

virtcontainers/noop_agent_test.go

@@ -56,6 +56,7 @@ func TestNoopAgentExec(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	if _, err = n.exec(pod, *container, cmd); err != nil {
 		t.Fatal(err)
@@ -88,6 +89,7 @@ func TestNoopAgentCreateContainer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	if err := n.startPod(*pod); err != nil {
 		t.Fatal(err)
@@ -104,6 +106,7 @@ func TestNoopAgentStartContainer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	err = n.startContainer(*pod, container)
 	if err != nil {
@@ -117,6 +120,7 @@ func TestNoopAgentStopContainer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	err = n.stopContainer(*pod, *container)
 	if err != nil {

virtcontainers/pod.go

@@ -642,23 +642,37 @@ func newPod(podConfig PodConfig) (*Pod, error) {
 		wg:              &sync.WaitGroup{},
 	}
 
-	if err := p.storage.createAllResources(*p); err != nil {
+	if err = globalPodList.addPod(p); err != nil {
 		return nil, err
 	}
 
-	if err := p.hypervisor.init(p); err != nil {
+	defer func() {
-		p.storage.deletePodResources(p.id, nil)
+		if err != nil {
+			p.Logger().WithError(err).WithField("podid", p.id).Error("Create new pod failed")
+			globalPodList.removePod(p.id)
+		}
+	}()
+
+	if err = p.storage.createAllResources(*p); err != nil {
 		return nil, err
 	}
 
-	if err := p.hypervisor.createPod(podConfig); err != nil {
+	defer func() {
-		p.storage.deletePodResources(p.id, nil)
+		if err != nil {
+			p.storage.deletePodResources(p.id, nil)
+		}
+	}()
+
+	if err = p.hypervisor.init(p); err != nil {
+		return nil, err
+	}
+
+	if err = p.hypervisor.createPod(podConfig); err != nil {
 		return nil, err
 	}
 
 	agentConfig := newAgentConfig(podConfig)
-	if err := p.agent.init(p, agentConfig); err != nil {
+	if err = p.agent.init(p, agentConfig); err != nil {
-		p.storage.deletePodResources(p.id, nil)
 		return nil, err
 	}
 
@@ -688,6 +702,11 @@ func fetchPod(podID string) (pod *Pod, err error) {
 		return nil, errNeedPodID
 	}
 
+	pod, err = globalPodList.lookupPod(podID)
+	if pod != nil && err == nil {
+		return pod, err
+	}
+
 	fs := filesystem{}
 	config, err := fs.fetchPodConfig(podID)
 	if err != nil {
@@ -766,6 +785,8 @@ func (p *Pod) delete() error {
 		}
 	}
 
+	globalPodList.removePod(p.id)
+
 	return p.storage.deletePodResources(p.id, nil)
 }
 

virtcontainers/pod_test.go

@@ -86,6 +86,7 @@ func TestCreateEmtpyPod(t *testing.T) {
 	if err == nil {
 		t.Fatalf("VirtContainers should not allow empty pods")
 	}
+	defer cleanUp()
 }
 
 func TestCreateEmtpyHypervisorPod(t *testing.T) {
@@ -93,6 +94,7 @@ func TestCreateEmtpyHypervisorPod(t *testing.T) {
 	if err == nil {
 		t.Fatalf("VirtContainers should not allow pods with empty hypervisors")
 	}
+	defer cleanUp()
 }
 
 func TestCreateMockPod(t *testing.T) {
@@ -102,6 +104,7 @@ func TestCreateMockPod(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 }
 
 func TestCreatePodEmtpyID(t *testing.T) {
@@ -111,6 +114,7 @@ func TestCreatePodEmtpyID(t *testing.T) {
 	if err == nil {
 		t.Fatalf("Expected pod with empty ID to fail, but got pod %v", p)
 	}
+	defer cleanUp()
 }
 
 func testPodStateTransition(t *testing.T, state stateString, newState stateString) error {
@@ -120,6 +124,7 @@ func testPodStateTransition(t *testing.T, state stateString, newState stateStrin
 	if err != nil {
 		return err
 	}
+	defer cleanUp()
 
 	p.state = State{
 		State: state,
@@ -537,6 +542,7 @@ func TestPodSetPodAndContainerState(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	l := len(p.GetAllContainers())
 	if l != 1 {
@@ -896,6 +902,7 @@ func TestPodGetContainer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	contID := "999"
 	contConfig := newTestContainerConfigNoop(contID)
@@ -941,6 +948,7 @@ func TestContainerSetStateBlockIndex(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	fs := &filesystem{}
 	pod.storage = fs
@@ -1039,6 +1047,7 @@ func TestContainerStateSetFstype(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanUp()
 
 	fs := &filesystem{}
 	pod.storage = fs

virtcontainers/podlist.go

@@ -0,0 +1,60 @@
+//
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// 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 virtcontainers
+
+import (
+	"fmt"
+	"sync"
+)
+
+type podList struct {
+	lock sync.RWMutex
+	pods map[string]*Pod
+}
+
+// globalPodList tracks pods globally
+var globalPodList = &podList{pods: make(map[string]*Pod)}
+
+func (p *podList) addPod(pod *Pod) (err error) {
+	if pod == nil {
+		return nil
+	}
+
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	if p.pods[pod.id] == nil {
+		p.pods[pod.id] = pod
+	} else {
+		err = fmt.Errorf("pod %s exists", pod.id)
+	}
+	return err
+}
+
+func (p *podList) removePod(id string) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	delete(p.pods, id)
+}
+
+func (p *podList) lookupPod(id string) (*Pod, error) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	if p.pods[id] != nil {
+		return p.pods[id], nil
+	}
+	return nil, fmt.Errorf("pod %s does not exist", id)
+}

virtcontainers/podlist_test.go

@@ -0,0 +1,42 @@
+//
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// 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 virtcontainers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPodListOperations(t *testing.T) {
+	p := &Pod{id: "testpodListpod"}
+	l := &podList{pods: make(map[string]*Pod)}
+	err := l.addPod(p)
+	assert.Nil(t, err, "addPod failed")
+
+	err = l.addPod(p)
+	assert.NotNil(t, err, "add same pod should fail")
+
+	np, err := l.lookupPod(p.id)
+	assert.Nil(t, err, "lookupPod failed")
+	assert.Equal(t, np, p, "lookupPod returns different pod %v:%v", np, p)
+
+	_, err = l.lookupPod("some-non-existing-pod-name")
+	assert.NotNil(t, err, "lookupPod for non-existing pod should fail")
+
+	l.removePod(p.id)
+}

virtcontainers/proxy.go

@@ -21,6 +21,7 @@ import (
 	"path/filepath"
 
 	"github.com/mitchellh/mapstructure"
+	"github.com/sirupsen/logrus"
 )
 
 // ProxyConfig is a structure storing information needed from any
@@ -34,6 +35,7 @@ type ProxyConfig struct {
 // for the execution of the proxy binary.
 type proxyParams struct {
 	agentURL string
+	logger   *logrus.Entry
 }
 
 // ProxyType describes a proxy type.
@@ -51,6 +53,9 @@ const (
 
 	// KataProxyType is the kataProxy.
 	KataProxyType ProxyType = "kataProxy"
+
+	// KataBuiltInProxyType is the kataBuiltInProxy.
+	KataBuiltInProxyType ProxyType = "kataBuiltInProxy"
 )
 
 const (
@@ -59,6 +64,14 @@ const (
 	waitForProxyTimeoutSecs = 5.0
 )
 
+const (
+	// unix socket type of console
+	consoleProtoUnix = "unix"
+
+	// pty type of console. Used mostly by kvmtools.
+	consoleProtoPty = "pty"
+)
+
 // Set sets a proxy type based on the input string.
 func (pType *ProxyType) Set(value string) error {
 	switch value {
@@ -74,6 +87,9 @@ func (pType *ProxyType) Set(value string) error {
 	case "kataProxy":
 		*pType = KataProxyType
 		return nil
+	case "kataBuiltInProxy":
+		*pType = KataBuiltInProxyType
+		return nil
 	default:
 		return fmt.Errorf("Unknown proxy type %s", value)
 	}
@@ -90,6 +106,8 @@ func (pType *ProxyType) String() string {
 		return string(CCProxyType)
 	case KataProxyType:
 		return string(KataProxyType)
+	case KataBuiltInProxyType:
+		return string(KataBuiltInProxyType)
 	default:
 		return ""
 	}
@@ -106,6 +124,8 @@ func newProxy(pType ProxyType) (proxy, error) {
 		return &ccProxy{}, nil
 	case KataProxyType:
 		return &kataProxy{}, nil
+	case KataBuiltInProxyType:
+		return &kataBuiltInProxy{}, nil
 	default:
 		return &noopProxy{}, nil
 	}
@@ -148,6 +168,10 @@ func defaultProxyURL(pod Pod, socketType string) (string, error) {
 	}
 }
 
+func isProxyBuiltIn(pType ProxyType) bool {
+	return pType == KataBuiltInProxyType
+}
+
 // proxy is the virtcontainers proxy interface.
 type proxy interface {
 	// start launches a proxy instance for the specified pod, returning

virtcontainers/proxy_test.go

@@ -52,6 +52,10 @@ func TestSetNoProxyType(t *testing.T) {
 	testSetProxyType(t, "noProxy", NoProxyType)
 }
 
+func TestSetKataBuiltInProxyType(t *testing.T) {
+	testSetProxyType(t, "kataBuiltInProxy", KataBuiltInProxyType)
+}
+
 func TestSetUnknownProxyType(t *testing.T) {
 	var proxyType ProxyType
 
@@ -97,6 +101,11 @@ func TestStringFromNoopProxyType(t *testing.T) {
 	testStringFromProxyType(t, proxyType, "noopProxy")
 }
 
+func TestStringFromKataBuiltInProxyType(t *testing.T) {
+	proxyType := KataBuiltInProxyType
+	testStringFromProxyType(t, proxyType, "kataBuiltInProxy")
+}
+
 func TestStringFromUnknownProxyType(t *testing.T) {
 	var proxyType ProxyType
 	testStringFromProxyType(t, proxyType, "")
@@ -137,6 +146,12 @@ func TestNewProxyFromNoopProxyType(t *testing.T) {
 	testNewProxyFromProxyType(t, proxyType, expectedProxy)
 }
 
+func TestNewProxyFromKataBuiltInProxyType(t *testing.T) {
+	proxyType := KataBuiltInProxyType
+	expectedProxy := &kataBuiltInProxy{}
+	testNewProxyFromProxyType(t, proxyType, expectedProxy)
+}
+
 func TestNewProxyFromUnknownProxyType(t *testing.T) {
 	var proxyType ProxyType
 

virtcontainers/shim.go

@@ -40,6 +40,9 @@ const (
 
 	// KataShimType is the Kata Containers shim type.
 	KataShimType ShimType = "kataShim"
+
+	// KataBuiltInShimType is the Kata Containers builtin shim type.
+	KataBuiltInShimType ShimType = "kataBuiltInShim"
 )
 
 var waitForShimTimeout = 10.0
@@ -71,16 +74,16 @@ func (pType *ShimType) Set(value string) error {
 	switch value {
 	case "noopShim":
 		*pType = NoopShimType
-		return nil
 	case "ccShim":
 		*pType = CCShimType
-		return nil
 	case "kataShim":
 		*pType = KataShimType
-		return nil
+	case "kataBuiltInShim":
+		*pType = KataBuiltInShimType
 	default:
 		return fmt.Errorf("Unknown shim type %s", value)
 	}
+	return nil
 }
 
 // String converts a shim type to a string.
@@ -92,6 +95,8 @@ func (pType *ShimType) String() string {
 		return string(CCShimType)
 	case KataShimType:
 		return string(KataShimType)
+	case KataBuiltInShimType:
+		return string(KataBuiltInShimType)
 	default:
 		return ""
 	}
@@ -106,6 +111,8 @@ func newShim(pType ShimType) (shim, error) {
 		return &ccShim{}, nil
 	case KataShimType:
 		return &kataShim{}, nil
+	case KataBuiltInShimType:
+		return &kataBuiltInShim{}, nil
 	default:
 		return &noopShim{}, nil
 	}
@@ -114,7 +121,7 @@ func newShim(pType ShimType) (shim, error) {
 // newShimConfig returns a shim config from a generic PodConfig interface.
 func newShimConfig(config PodConfig) interface{} {
 	switch config.ShimType {
-	case NoopShimType:
+	case NoopShimType, KataBuiltInShimType:
 		return nil
 	case CCShimType, KataShimType:
 		var shimConfig ShimConfig
@@ -147,6 +154,10 @@ func signalShim(pid int, sig syscall.Signal) error {
 }
 
 func stopShim(pid int) error {
+	if pid <= 0 {
+		return nil
+	}
+
 	if err := signalShim(pid, syscall.SIGKILL); err != nil && err != syscall.ESRCH {
 		return err
 	}
@@ -233,6 +244,10 @@ func startShim(args []string, params ShimParams) (int, error) {
 }
 
 func isShimRunning(pid int) (bool, error) {
+	if pid <= 0 {
+		return false, nil
+	}
+
 	process, err := os.FindProcess(pid)
 	if err != nil {
 		return false, err

virtcontainers/shim_test.go

@@ -89,6 +89,11 @@ func TestStringFromNoopShimType(t *testing.T) {
 	testStringFromShimType(t, shimType, "noopShim")
 }
 
+func TestStringFromKataBuiltInShimType(t *testing.T) {
+	shimType := KataBuiltInShimType
+	testStringFromShimType(t, shimType, "kataBuiltInShim")
+}
+
 func TestStringFromUnknownShimType(t *testing.T) {
 	var shimType ShimType
 	testStringFromShimType(t, shimType, "")
@@ -123,6 +128,12 @@ func TestNewShimFromNoopShimType(t *testing.T) {
 	testNewShimFromShimType(t, shimType, expectedShim)
 }
 
+func TestNewShimFromKataBuiltInShimType(t *testing.T) {
+	shimType := KataBuiltInShimType
+	expectedShim := &kataBuiltInShim{}
+	testNewShimFromShimType(t, shimType, expectedShim)
+}
+
 func TestNewShimFromUnknownShimType(t *testing.T) {
 	var shimType ShimType
 
@@ -170,6 +181,14 @@ func TestNewShimConfigFromNoopShimPodConfig(t *testing.T) {
 	testNewShimConfigFromPodConfig(t, podConfig, nil)
 }
 
+func TestNewShimConfigFromKataBuiltInShimPodConfig(t *testing.T) {
+	podConfig := PodConfig{
+		ShimType: KataBuiltInShimType,
+	}
+
+	testNewShimConfigFromPodConfig(t, podConfig, nil)
+}
+
 func TestNewShimConfigFromUnknownShimPodConfig(t *testing.T) {
 	var shimType ShimType
 

virtcontainers/virtcontainers_test.go

@@ -55,6 +55,7 @@ var testHyperstartTtySocket = ""
 // cleanUp Removes any stale pod/container state that can affect
 // the next test to run.
 func cleanUp() {
+	globalPodList.removePod(testPodID)
 	for _, dir := range []string{testDir, defaultSharedDir} {
 		os.RemoveAll(dir)
 		os.MkdirAll(dir, dirMode)