-
Notifications
You must be signed in to change notification settings - Fork 7
/
portforward.go
82 lines (70 loc) · 2.28 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
// Copyright 2024 Edgeless Systems GmbH
// SPDX-License-Identifier: AGPL-3.0-only
package kubeclient
import (
"context"
"fmt"
"net/http"
"k8s.io/client-go/tools/portforward"
"k8s.io/client-go/transport/spdy"
)
// PortForwardPod starts a port forward to the selected pod.
//
// On success, the function returns a TCP address that clients can connect to and a function to
// cancel the port forwarding.
func (k *Kubeclient) PortForwardPod(ctx context.Context, namespace, podName, remotePort string) (string, func(), error) {
// We can only forward to the pod once it's ready.
if err := k.WaitForPod(ctx, namespace, podName); err != nil {
return "", nil, fmt.Errorf("waiting for pod %s: %w", podName, err)
}
// This channel sends a stop request to the portforwarding goroutine.
stopCh := make(chan struct{}, 1)
// The portforwarding goroutine closes this channel when it's ready.
readyCh := make(chan struct{})
// Any error returned by the background port-forwarder is sent to this channel.
errorCh := make(chan error)
// Ports are forwarded by upgrading this POST request to a SPDY connection.
req := k.client.CoreV1().RESTClient().Post().
Resource("pods").
Namespace(namespace).
Name(podName).
SubResource("portforward")
transport, upgrader, err := spdy.RoundTripperFor(k.config)
if err != nil {
return "", nil, fmt.Errorf("creating round tripper: %w", err)
}
dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, req.URL())
fw, err := portforward.NewOnAddresses(
dialer,
[]string{"localhost"},
[]string{fmt.Sprintf("0:%s", remotePort)},
stopCh, readyCh,
nil, nil,
)
if err != nil {
return "", nil, fmt.Errorf("creating portforwarder: %w", err)
}
go func() {
if err := fw.ForwardPorts(); err != nil {
errorCh <- err
}
}()
select {
case <-readyCh:
ports, err := fw.GetPorts()
if err != nil {
close(stopCh)
return "", nil, fmt.Errorf("getting ports: %w", err)
}
cleanUp := func() {
close(stopCh)
}
return fmt.Sprintf("localhost:%d", ports[0].Local), cleanUp, nil
case <-ctx.Done():
close(stopCh)
return "", nil, fmt.Errorf("waiting for port forward to be ready: %w", ctx.Err())
case err := <-errorCh:
close(stopCh)
return "", nil, fmt.Errorf("background port-forwarding routine failed: %w", err)
}
}