connection.go

Documentation: k8s.io/apimachinery/pkg/util/httpstream/spdy

     1  /*
     2  Copyright 2015 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package spdy
    18  
    19  import (
    20  	"net"
    21  	"net/http"
    22  	"sync"
    23  	"time"
    24  
    25  	"github.com/moby/spdystream"
    26  	"k8s.io/apimachinery/pkg/util/httpstream"
    27  	"k8s.io/klog/v2"
    28  )
    29  
    30  // connection maintains state about a spdystream.Connection and its associated
    31  // streams.
    32  type connection struct {
    33  	conn             *spdystream.Connection
    34  	streams          map[uint32]httpstream.Stream
    35  	streamLock       sync.Mutex
    36  	newStreamHandler httpstream.NewStreamHandler
    37  	ping             func() (time.Duration, error)
    38  }
    39  
    40  // NewClientConnection creates a new SPDY client connection.
    41  func NewClientConnection(conn net.Conn) (httpstream.Connection, error) {
    42  	return NewClientConnectionWithPings(conn, 0)
    43  }
    44  
    45  // NewClientConnectionWithPings creates a new SPDY client connection.
    46  //
    47  // If pingPeriod is non-zero, a background goroutine will send periodic Ping
    48  // frames to the server. Use this to keep idle connections through certain load
    49  // balancers alive longer.
    50  func NewClientConnectionWithPings(conn net.Conn, pingPeriod time.Duration) (httpstream.Connection, error) {
    51  	spdyConn, err := spdystream.NewConnection(conn, false)
    52  	if err != nil {
    53  		defer conn.Close()
    54  		return nil, err
    55  	}
    56  
    57  	return newConnection(spdyConn, httpstream.NoOpNewStreamHandler, pingPeriod, spdyConn.Ping), nil
    58  }
    59  
    60  // NewServerConnection creates a new SPDY server connection. newStreamHandler
    61  // will be invoked when the server receives a newly created stream from the
    62  // client.
    63  func NewServerConnection(conn net.Conn, newStreamHandler httpstream.NewStreamHandler) (httpstream.Connection, error) {
    64  	return NewServerConnectionWithPings(conn, newStreamHandler, 0)
    65  }
    66  
    67  // NewServerConnectionWithPings creates a new SPDY server connection.
    68  // newStreamHandler will be invoked when the server receives a newly created
    69  // stream from the client.
    70  //
    71  // If pingPeriod is non-zero, a background goroutine will send periodic Ping
    72  // frames to the server. Use this to keep idle connections through certain load
    73  // balancers alive longer.
    74  func NewServerConnectionWithPings(conn net.Conn, newStreamHandler httpstream.NewStreamHandler, pingPeriod time.Duration) (httpstream.Connection, error) {
    75  	spdyConn, err := spdystream.NewConnection(conn, true)
    76  	if err != nil {
    77  		defer conn.Close()
    78  		return nil, err
    79  	}
    80  
    81  	return newConnection(spdyConn, newStreamHandler, pingPeriod, spdyConn.Ping), nil
    82  }
    83  
    84  // newConnection returns a new connection wrapping conn. newStreamHandler
    85  // will be invoked when the server receives a newly created stream from the
    86  // client.
    87  func newConnection(conn *spdystream.Connection, newStreamHandler httpstream.NewStreamHandler, pingPeriod time.Duration, pingFn func() (time.Duration, error)) httpstream.Connection {
    88  	c := &connection{
    89  		conn:             conn,
    90  		newStreamHandler: newStreamHandler,
    91  		ping:             pingFn,
    92  		streams:          make(map[uint32]httpstream.Stream),
    93  	}
    94  	go conn.Serve(c.newSpdyStream)
    95  	if pingPeriod > 0 && pingFn != nil {
    96  		go c.sendPings(pingPeriod)
    97  	}
    98  	return c
    99  }
   100  
   101  // createStreamResponseTimeout indicates how long to wait for the other side to
   102  // acknowledge the new stream before timing out.
   103  const createStreamResponseTimeout = 30 * time.Second
   104  
   105  // Close first sends a reset for all of the connection's streams, and then
   106  // closes the underlying spdystream.Connection.
   107  func (c *connection) Close() error {
   108  	c.streamLock.Lock()
   109  	for _, s := range c.streams {
   110  		// calling Reset instead of Close ensures that all streams are fully torn down
   111  		s.Reset()
   112  	}
   113  	c.streams = make(map[uint32]httpstream.Stream, 0)
   114  	c.streamLock.Unlock()
   115  
   116  	// now that all streams are fully torn down, it's safe to call close on the underlying connection,
   117  	// which should be able to terminate immediately at this point, instead of waiting for any
   118  	// remaining graceful stream termination.
   119  	return c.conn.Close()
   120  }
   121  
   122  // RemoveStreams can be used to removes a set of streams from the Connection.
   123  func (c *connection) RemoveStreams(streams ...httpstream.Stream) {
   124  	c.streamLock.Lock()
   125  	for _, stream := range streams {
   126  		// It may be possible that the provided stream is nil if timed out.
   127  		if stream != nil {
   128  			delete(c.streams, stream.Identifier())
   129  		}
   130  	}
   131  	c.streamLock.Unlock()
   132  }
   133  
   134  // CreateStream creates a new stream with the specified headers and registers
   135  // it with the connection.
   136  func (c *connection) CreateStream(headers http.Header) (httpstream.Stream, error) {
   137  	stream, err := c.conn.CreateStream(headers, nil, false)
   138  	if err != nil {
   139  		return nil, err
   140  	}
   141  	if err = stream.WaitTimeout(createStreamResponseTimeout); err != nil {
   142  		return nil, err
   143  	}
   144  
   145  	c.registerStream(stream)
   146  	return stream, nil
   147  }
   148  
   149  // registerStream adds the stream s to the connection's list of streams that
   150  // it owns.
   151  func (c *connection) registerStream(s httpstream.Stream) {
   152  	c.streamLock.Lock()
   153  	c.streams[s.Identifier()] = s
   154  	c.streamLock.Unlock()
   155  }
   156  
   157  // CloseChan returns a channel that, when closed, indicates that the underlying
   158  // spdystream.Connection has been closed.
   159  func (c *connection) CloseChan() <-chan bool {
   160  	return c.conn.CloseChan()
   161  }
   162  
   163  // newSpdyStream is the internal new stream handler used by spdystream.Connection.Serve.
   164  // It calls connection's newStreamHandler, giving it the opportunity to accept or reject
   165  // the stream. If newStreamHandler returns an error, the stream is rejected. If not, the
   166  // stream is accepted and registered with the connection.
   167  func (c *connection) newSpdyStream(stream *spdystream.Stream) {
   168  	replySent := make(chan struct{})
   169  	err := c.newStreamHandler(stream, replySent)
   170  	rejectStream := (err != nil)
   171  	if rejectStream {
   172  		klog.Warningf("Stream rejected: %v", err)
   173  		stream.Reset()
   174  		return
   175  	}
   176  
   177  	c.registerStream(stream)
   178  	stream.SendReply(http.Header{}, rejectStream)
   179  	close(replySent)
   180  }
   181  
   182  // SetIdleTimeout sets the amount of time the connection may remain idle before
   183  // it is automatically closed.
   184  func (c *connection) SetIdleTimeout(timeout time.Duration) {
   185  	c.conn.SetIdleTimeout(timeout)
   186  }
   187  
   188  func (c *connection) sendPings(period time.Duration) {
   189  	t := time.NewTicker(period)
   190  	defer t.Stop()
   191  	for {
   192  		select {
   193  		case <-c.conn.CloseChan():
   194  			return
   195  		case <-t.C:
   196  		}
   197  		if _, err := c.ping(); err != nil {
   198  			klog.V(3).Infof("SPDY Ping failed: %v", err)
   199  			// Continue, in case this is a transient failure.
   200  			// c.conn.CloseChan above will tell us when the connection is
   201  			// actually closed.
   202  		}
   203  	}
   204  }
   205
View as plain text