rebase: upgrade snapshot v1beta1 api and clientset to v1 version

At present e2e snapshot code make use of snapshot v1beta1 api and client
This patch address the same and snapshot api clientset/apis are updated
to v1 version.

Signed-off-by: Humble Chirammal <hchiramm@redhat.com>

vendor/google.golang.org/grpc/internal/transport/controlbuf.go

@@ -20,17 +20,13 @@ package transport
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"runtime"
-	"strconv"
 	"sync"
 	"sync/atomic"
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
-	"google.golang.org/grpc/internal/grpcutil"
-	"google.golang.org/grpc/status"
 )
 
 var updateHeaderTblSize = func(e *hpack.Encoder, v uint32) {
@@ -132,14 +128,6 @@ type cleanupStream struct {
 
 func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM
 
-type earlyAbortStream struct {
-	streamID       uint32
-	contentSubtype string
-	status         *status.Status
-}
-
-func (*earlyAbortStream) isTransportResponseFrame() bool { return false }
-
 type dataFrame struct {
 	streamID  uint32
 	endStream bool
@@ -761,24 +749,6 @@ func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error {
 	return nil
 }
 
-func (l *loopyWriter) earlyAbortStreamHandler(eas *earlyAbortStream) error {
-	if l.side == clientSide {
-		return errors.New("earlyAbortStream not handled on client")
-	}
-
-	headerFields := []hpack.HeaderField{
-		{Name: ":status", Value: "200"},
-		{Name: "content-type", Value: grpcutil.ContentType(eas.contentSubtype)},
-		{Name: "grpc-status", Value: strconv.Itoa(int(eas.status.Code()))},
-		{Name: "grpc-message", Value: encodeGrpcMessage(eas.status.Message())},
-	}
-
-	if err := l.writeHeader(eas.streamID, true, headerFields, nil); err != nil {
-		return err
-	}
-	return nil
-}
-
 func (l *loopyWriter) incomingGoAwayHandler(*incomingGoAway) error {
 	if l.side == clientSide {
 		l.draining = true
@@ -817,8 +787,6 @@ func (l *loopyWriter) handle(i interface{}) error {
 		return l.registerStreamHandler(i)
 	case *cleanupStream:
 		return l.cleanupStreamHandler(i)
-	case *earlyAbortStream:
-		return l.earlyAbortStreamHandler(i)
 	case *incomingGoAway:
 		return l.incomingGoAwayHandler(i)
 	case *dataFrame:

vendor/google.golang.org/grpc/internal/transport/http2_client.go

@@ -32,14 +32,15 @@ import (
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/internal/channelz"
-	icredentials "google.golang.org/grpc/internal/credentials"
 	"google.golang.org/grpc/internal/grpcutil"
 	imetadata "google.golang.org/grpc/internal/metadata"
-	"google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/internal/transport/networktype"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/channelz"
+	"google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
@@ -115,9 +116,6 @@ type http2Client struct {
 	// goAwayReason records the http2.ErrCode and debug data received with the
 	// GoAway frame.
 	goAwayReason GoAwayReason
-	// goAwayDebugMessage contains a detailed human readable string about a
-	// GoAway frame, useful for error messages.
-	goAwayDebugMessage string
 	// A condition variable used to signal when the keepalive goroutine should
 	// go dormant. The condition for dormancy is based on the number of active
 	// streams and the `PermitWithoutStream` keepalive client parameter. And
@@ -240,7 +238,8 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 		// Attributes field of resolver.Address, which is shoved into connectCtx
 		// and passed to the credential handshaker. This makes it possible for
 		// address specific arbitrary data to reach the credential handshaker.
-		connectCtx = icredentials.NewClientHandshakeInfoContext(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
+		contextWithHandshakeInfo := internal.NewClientHandshakeInfoContext.(func(context.Context, credentials.ClientHandshakeInfo) context.Context)
+		connectCtx = contextWithHandshakeInfo(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
 		conn, authInfo, err = transportCreds.ClientHandshake(connectCtx, addr.ServerName, conn)
 		if err != nil {
 			return nil, connectionErrorf(isTemporary(err), err, "transport: authentication handshake failed: %v", err)
@@ -348,14 +347,12 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	// Send connection preface to server.
 	n, err := t.conn.Write(clientPreface)
 	if err != nil {
-		err = connectionErrorf(true, err, "transport: failed to write client preface: %v", err)
-		t.Close(err)
-		return nil, err
+		t.Close()
+		return nil, connectionErrorf(true, err, "transport: failed to write client preface: %v", err)
 	}
 	if n != len(clientPreface) {
-		err = connectionErrorf(true, nil, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface))
-		t.Close(err)
-		return nil, err
+		t.Close()
+		return nil, connectionErrorf(true, err, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface))
 	}
 	var ss []http2.Setting
 
@@ -373,16 +370,14 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	}
 	err = t.framer.fr.WriteSettings(ss...)
 	if err != nil {
-		err = connectionErrorf(true, err, "transport: failed to write initial settings frame: %v", err)
-		t.Close(err)
-		return nil, err
+		t.Close()
+		return nil, connectionErrorf(true, err, "transport: failed to write initial settings frame: %v", err)
 	}
 	// Adjust the connection flow control window if needed.
 	if delta := uint32(icwz - defaultWindowSize); delta > 0 {
 		if err := t.framer.fr.WriteWindowUpdate(0, delta); err != nil {
-			err = connectionErrorf(true, err, "transport: failed to write window update: %v", err)
-			t.Close(err)
-			return nil, err
+			t.Close()
+			return nil, connectionErrorf(true, err, "transport: failed to write window update: %v", err)
 		}
 	}
 
@@ -419,7 +414,6 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream {
 		buf:            newRecvBuffer(),
 		headerChan:     make(chan struct{}),
 		contentSubtype: callHdr.ContentSubtype,
-		doneFunc:       callHdr.DoneFunc,
 	}
 	s.wq = newWriteQuota(defaultWriteQuota, s.done)
 	s.requestRead = func(n int) {
@@ -459,7 +453,7 @@ func (t *http2Client) createHeaderFields(ctx context.Context, callHdr *CallHdr)
 		Method:   callHdr.Method,
 		AuthInfo: t.authInfo,
 	}
-	ctxWithRequestInfo := icredentials.NewRequestInfoContext(ctx, ri)
+	ctxWithRequestInfo := internal.NewRequestInfoContext.(func(context.Context, credentials.RequestInfo) context.Context)(ctx, ri)
 	authData, err := t.getTrAuthData(ctxWithRequestInfo, aud)
 	if err != nil {
 		return nil, err
@@ -838,9 +832,6 @@ func (t *http2Client) closeStream(s *Stream, err error, rst bool, rstCode http2.
 	t.controlBuf.executeAndPut(addBackStreamQuota, cleanup)
 	// This will unblock write.
 	close(s.done)
-	if s.doneFunc != nil {
-		s.doneFunc()
-	}
 }
 
 // Close kicks off the shutdown process of the transport. This should be called
@@ -850,12 +841,12 @@ func (t *http2Client) closeStream(s *Stream, err error, rst bool, rstCode http2.
 // This method blocks until the addrConn that initiated this transport is
 // re-connected. This happens because t.onClose() begins reconnect logic at the
 // addrConn level and blocks until the addrConn is successfully connected.
-func (t *http2Client) Close(err error) {
+func (t *http2Client) Close() error {
 	t.mu.Lock()
 	// Make sure we only Close once.
 	if t.state == closing {
 		t.mu.Unlock()
-		return
+		return nil
 	}
 	// Call t.onClose before setting the state to closing to prevent the client
 	// from attempting to create new streams ASAP.
@@ -871,19 +862,13 @@ func (t *http2Client) Close(err error) {
 	t.mu.Unlock()
 	t.controlBuf.finish()
 	t.cancel()
-	t.conn.Close()
+	err := t.conn.Close()
 	if channelz.IsOn() {
 		channelz.RemoveEntry(t.channelzID)
 	}
-	// Append info about previous goaways if there were any, since this may be important
-	// for understanding the root cause for this connection to be closed.
-	_, goAwayDebugMessage := t.GetGoAwayReason()
-	if len(goAwayDebugMessage) > 0 {
-		err = fmt.Errorf("closing transport due to: %v, received prior goaway: %v", err, goAwayDebugMessage)
-	}
 	// Notify all active streams.
 	for _, s := range streams {
-		t.closeStream(s, err, false, http2.ErrCodeNo, status.New(codes.Unavailable, err.Error()), nil, false)
+		t.closeStream(s, ErrConnClosing, false, http2.ErrCodeNo, status.New(codes.Unavailable, ErrConnClosing.Desc), nil, false)
 	}
 	if t.statsHandler != nil {
 		connEnd := &stats.ConnEnd{
@@ -891,6 +876,7 @@ func (t *http2Client) Close(err error) {
 		}
 		t.statsHandler.HandleConn(t.ctx, connEnd)
 	}
+	return err
 }
 
 // GracefulClose sets the state to draining, which prevents new streams from
@@ -909,7 +895,7 @@ func (t *http2Client) GracefulClose() {
 	active := len(t.activeStreams)
 	t.mu.Unlock()
 	if active == 0 {
-		t.Close(ErrConnClosing)
+		t.Close()
 		return
 	}
 	t.controlBuf.put(&incomingGoAway{})
@@ -1155,9 +1141,9 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 		}
 	}
 	id := f.LastStreamID
-	if id > 0 && id%2 == 0 {
+	if id > 0 && id%2 != 1 {
 		t.mu.Unlock()
-		t.Close(connectionErrorf(true, nil, "received goaway with non-zero even-numbered numbered stream id: %v", id))
+		t.Close()
 		return
 	}
 	// A client can receive multiple GoAways from the server (see
@@ -1175,7 +1161,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 		// If there are multiple GoAways the first one should always have an ID greater than the following ones.
 		if id > t.prevGoAwayID {
 			t.mu.Unlock()
-			t.Close(connectionErrorf(true, nil, "received goaway with stream id: %v, which exceeds stream id of previous goaway: %v", id, t.prevGoAwayID))
+			t.Close()
 			return
 		}
 	default:
@@ -1205,7 +1191,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 	active := len(t.activeStreams)
 	t.mu.Unlock()
 	if active == 0 {
-		t.Close(connectionErrorf(true, nil, "received goaway and there are no active streams"))
+		t.Close()
 	}
 }
 
@@ -1221,13 +1207,12 @@ func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) {
 			t.goAwayReason = GoAwayTooManyPings
 		}
 	}
-	t.goAwayDebugMessage = fmt.Sprintf("code: %s, debug data: %v", f.ErrCode, string(f.DebugData()))
 }
 
-func (t *http2Client) GetGoAwayReason() (GoAwayReason, string) {
+func (t *http2Client) GetGoAwayReason() GoAwayReason {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	return t.goAwayReason, t.goAwayDebugMessage
+	return t.goAwayReason
 }
 
 func (t *http2Client) handleWindowUpdate(f *http2.WindowUpdateFrame) {
@@ -1324,8 +1309,7 @@ func (t *http2Client) reader() {
 	// Check the validity of server preface.
 	frame, err := t.framer.fr.ReadFrame()
 	if err != nil {
-		err = connectionErrorf(true, err, "error reading server preface: %v", err)
-		t.Close(err) // this kicks off resetTransport, so must be last before return
+		t.Close() // this kicks off resetTransport, so must be last before return
 		return
 	}
 	t.conn.SetReadDeadline(time.Time{}) // reset deadline once we get the settings frame (we didn't time out, yay!)
@@ -1334,8 +1318,7 @@ func (t *http2Client) reader() {
 	}
 	sf, ok := frame.(*http2.SettingsFrame)
 	if !ok {
-		// this kicks off resetTransport, so must be last before return
-		t.Close(connectionErrorf(true, nil, "initial http2 frame from server is not a settings frame: %T", frame))
+		t.Close() // this kicks off resetTransport, so must be last before return
 		return
 	}
 	t.onPrefaceReceipt()
@@ -1371,7 +1354,7 @@ func (t *http2Client) reader() {
 				continue
 			} else {
 				// Transport error.
-				t.Close(connectionErrorf(true, err, "error reading from server: %v", err))
+				t.Close()
 				return
 			}
 		}
@@ -1430,7 +1413,7 @@ func (t *http2Client) keepalive() {
 				continue
 			}
 			if outstandingPing && timeoutLeft <= 0 {
-				t.Close(connectionErrorf(true, nil, "keepalive ping failed to receive ACK within timeout"))
+				t.Close()
 				return
 			}
 			t.mu.Lock()

vendor/google.golang.org/grpc/internal/transport/http2_server.go

@@ -26,7 +26,6 @@ import (
 	"io"
 	"math"
 	"net"
-	"net/http"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -356,6 +355,26 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	if state.data.statsTrace != nil {
 		s.ctx = stats.SetIncomingTrace(s.ctx, state.data.statsTrace)
 	}
+	if t.inTapHandle != nil {
+		var err error
+		info := &tap.Info{
+			FullMethodName: state.data.method,
+		}
+		s.ctx, err = t.inTapHandle(s.ctx, info)
+		if err != nil {
+			if logger.V(logLevel) {
+				logger.Warningf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
+			}
+			t.controlBuf.put(&cleanupStream{
+				streamID: s.id,
+				rst:      true,
+				rstCode:  http2.ErrCodeRefusedStream,
+				onWrite:  func() {},
+			})
+			s.cancel()
+			return false
+		}
+	}
 	t.mu.Lock()
 	if t.state != reachable {
 		t.mu.Unlock()
@@ -383,39 +402,6 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		return true
 	}
 	t.maxStreamID = streamID
-	if state.data.httpMethod != http.MethodPost {
-		t.mu.Unlock()
-		if logger.V(logLevel) {
-			logger.Warningf("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", state.data.httpMethod)
-		}
-		t.controlBuf.put(&cleanupStream{
-			streamID: streamID,
-			rst:      true,
-			rstCode:  http2.ErrCodeProtocol,
-			onWrite:  func() {},
-		})
-		s.cancel()
-		return false
-	}
-	if t.inTapHandle != nil {
-		var err error
-		if s.ctx, err = t.inTapHandle(s.ctx, &tap.Info{FullMethodName: state.data.method}); err != nil {
-			t.mu.Unlock()
-			if logger.V(logLevel) {
-				logger.Infof("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
-			}
-			stat, ok := status.FromError(err)
-			if !ok {
-				stat = status.New(codes.PermissionDenied, err.Error())
-			}
-			t.controlBuf.put(&earlyAbortStream{
-				streamID:       s.id,
-				contentSubtype: s.contentSubtype,
-				status:         stat,
-			})
-			return false
-		}
-	}
 	t.activeStreams[streamID] = s
 	if len(t.activeStreams) == 1 {
 		t.idle = time.Time{}

vendor/google.golang.org/grpc/internal/transport/http_util.go

@@ -111,7 +111,6 @@ type parsedHeaderData struct {
 	timeoutSet bool
 	timeout    time.Duration
 	method     string
-	httpMethod string
 	// key-value metadata map from the peer.
 	mdata          map[string][]string
 	statsTags      []byte
@@ -364,8 +363,6 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) {
 		}
 		d.data.statsTrace = v
 		d.addMetadata(f.Name, string(v))
-	case ":method":
-		d.data.httpMethod = f.Value
 	default:
 		if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
 			break

vendor/google.golang.org/grpc/internal/transport/transport.go

@@ -241,7 +241,6 @@ type Stream struct {
 	ctx          context.Context    // the associated context of the stream
 	cancel       context.CancelFunc // always nil for client side Stream
 	done         chan struct{}      // closed at the end of stream to unblock writers. On the client side.
-	doneFunc     func()             // invoked at the end of stream on client side.
 	ctxDone      <-chan struct{}    // same as done chan but for server side. Cache of ctx.Done() (for performance)
 	method       string             // the associated RPC method of the stream
 	recvCompress string
@@ -612,8 +611,6 @@ type CallHdr struct {
 	ContentSubtype string
 
 	PreviousAttempts int // value of grpc-previous-rpc-attempts header to set
-
-	DoneFunc func() // called when the stream is finished
 }
 
 // ClientTransport is the common interface for all gRPC client-side transport
@@ -622,7 +619,7 @@ type ClientTransport interface {
 	// Close tears down this transport. Once it returns, the transport
 	// should not be accessed any more. The caller must make sure this
 	// is called only once.
-	Close(err error)
+	Close() error
 
 	// GracefulClose starts to tear down the transport: the transport will stop
 	// accepting new RPCs and NewStream will return error. Once all streams are
@@ -656,9 +653,8 @@ type ClientTransport interface {
 	// HTTP/2).
 	GoAway() <-chan struct{}
 
-	// GetGoAwayReason returns the reason why GoAway frame was received, along
-	// with a human readable string with debug info.
-	GetGoAwayReason() (GoAwayReason, string)
+	// GetGoAwayReason returns the reason why GoAway frame was received.
+	GetGoAwayReason() GoAwayReason
 
 	// RemoteAddr returns the remote network address.
 	RemoteAddr() net.Addr