-
Notifications
You must be signed in to change notification settings - Fork 290
/
portforward.go
170 lines (146 loc) · 4.56 KB
/
portforward.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
package k8s
import (
"context"
"fmt"
"net"
"net/http"
"strconv"
v1 "k8s.io/client-go/kubernetes/typed/core/v1"
_ "k8s.io/client-go/plugin/pkg/client/auth/gcp" // registers gcp auth provider
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/portforward"
"k8s.io/client-go/transport/spdy"
"github.com/windmilleng/tilt/pkg/logger"
"github.com/pkg/errors"
)
type PortForwardClient interface {
// Creates a new port-forwarder that's bound to the given context's lifecycle.
// When the context is canceled, the port-forwarder will close.
CreatePortForwarder(ctx context.Context, namespace Namespace, podID PodID, localPort int, remotePort int, host string) (PortForwarder, error)
}
type PortForwarder interface {
// The local port we're listening on.
LocalPort() int
// Listens on the configured port and forward all traffic to the container.
// Returns when the port-forwarder sees an unrecoverable error or
// when the context passed at creation is canceled.
ForwardPorts() error
}
type portForwarder struct {
*portforward.PortForwarder
localPort int
}
func (pf portForwarder) LocalPort() int {
return pf.localPort
}
func (k K8sClient) CreatePortForwarder(ctx context.Context, namespace Namespace, podID PodID, optionalLocalPort, remotePort int, host string) (PortForwarder, error) {
localPort := optionalLocalPort
if localPort == 0 {
// preferably, we'd set the localport to 0, and let the underlying function pick a port for us,
// to avoid the race condition potential of something else grabbing this port between
// the call to `getAvailablePort` and whenever `portForwarder` actually binds the port.
// the k8s client supports a local port of 0, and stores the actual local port assigned in a field,
// but unfortunately does not export that field, so there is no way for the caller to know which
// local port to talk to.
var err error
localPort, err = getAvailablePort()
if err != nil {
return nil, errors.Wrap(err, "failed to find an available local port")
}
}
return k.portForwardClient.CreatePortForwarder(ctx, namespace, podID, localPort, remotePort, host)
}
type portForwardClient struct {
config *rest.Config
core v1.CoreV1Interface
}
func ProvidePortForwardClient(
maybeRESTConfig RESTConfigOrError,
maybeClientset ClientsetOrError) PortForwardClient {
if maybeRESTConfig.Error != nil {
return explodingPortForwardClient{error: maybeRESTConfig.Error}
}
if maybeClientset.Error != nil {
return explodingPortForwardClient{error: maybeClientset.Error}
}
return portForwardClient{
maybeRESTConfig.Config,
maybeClientset.Clientset.CoreV1(),
}
}
func (c portForwardClient) CreatePortForwarder(ctx context.Context, namespace Namespace, podID PodID, localPort int, remotePort int, host string) (PortForwarder, error) {
transport, upgrader, err := spdy.RoundTripperFor(c.config)
if err != nil {
return nil, errors.Wrap(err, "error getting roundtripper")
}
req := c.core.RESTClient().Post().
Resource("pods").
Namespace(namespace.String()).
Name(podID.String()).
SubResource("portforward")
dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", req.URL())
if err != nil {
return nil, errors.Wrap(err, "error creating dialer")
}
stopChan := make(chan struct{}, 1)
readyChan := make(chan struct{}, 1)
ports := []string{fmt.Sprintf("%d:%d", localPort, remotePort)}
var pf *portforward.PortForwarder
if host == "" {
pf, err = portforward.New(
dialer,
ports,
stopChan,
readyChan,
logger.Get(ctx).Writer(logger.DebugLvl),
logger.Get(ctx).Writer(logger.DebugLvl))
} else {
addresses := []string{host}
pf, err = portforward.NewOnAddresses(
dialer,
addresses,
ports,
stopChan,
readyChan,
logger.Get(ctx).Writer(logger.DebugLvl),
logger.Get(ctx).Writer(logger.DebugLvl))
}
if err != nil {
return nil, errors.Wrap(err, "error forwarding port")
}
go func() {
<-ctx.Done()
close(stopChan)
}()
return portForwarder{
PortForwarder: pf,
localPort: localPort,
}, nil
}
func getAvailablePort() (int, error) {
l, err := net.Listen("tcp", ":0")
if err != nil {
return 0, err
}
defer func() {
e := l.Close()
if err == nil {
err = e
}
}()
_, p, err := net.SplitHostPort(l.Addr().String())
if err != nil {
return 0, err
}
port, err := strconv.Atoi(p)
if err != nil {
return 0, err
}
return port, err
}
type explodingPortForwardClient struct {
error error
}
func (c explodingPortForwardClient) CreatePortForwarder(ctx context.Context, namespace Namespace, podID PodID, localPort int, remotePort int, host string) (PortForwarder, error) {
return nil, c.error
}