/
server.go
127 lines (109 loc) · 3.26 KB
/
server.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
package destination
import (
"fmt"
"net"
"strconv"
"strings"
pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
"github.com/linkerd/linkerd2/controller/k8s"
"github.com/linkerd/linkerd2/pkg/prometheus"
log "github.com/sirupsen/logrus"
"google.golang.org/grpc"
)
type server struct {
k8sAPI *k8s.API
resolvers []streamingDestinationResolver
enableTLS bool
}
// The Destination service serves service discovery information to the proxy.
// This implementation supports the "k8s" destination scheme and expects
// destination paths to be of the form:
// <service>.<namespace>.svc.cluster.local:<port>
//
// If the port is omitted, 80 is used as a default. If the namespace is
// omitted, "default" is used as a default.append
//
// Addresses for the given destination are fetched from the Kubernetes Endpoints
// API.
func NewServer(addr, k8sDNSZone string, enableTLS bool, k8sAPI *k8s.API, done chan struct{}) (*grpc.Server, net.Listener, error) {
resolvers, err := buildResolversList(k8sDNSZone, k8sAPI)
if err != nil {
return nil, nil, err
}
srv := server{
k8sAPI: k8sAPI,
resolvers: resolvers,
enableTLS: enableTLS,
}
lis, err := net.Listen("tcp", addr)
if err != nil {
return nil, nil, err
}
s := prometheus.NewGrpcServer()
pb.RegisterDestinationServer(s, &srv)
go func() {
<-done
for _, resolver := range resolvers {
resolver.stop()
}
}()
return s, lis, nil
}
func (s *server) Get(dest *pb.GetDestination, stream pb.Destination_GetServer) error {
log.Debugf("Get %v", dest)
if dest.Scheme != "k8s" {
err := fmt.Errorf("Unsupported scheme %v", dest.Scheme)
log.Error(err)
return err
}
hostPort := strings.Split(dest.Path, ":")
if len(hostPort) > 2 {
err := fmt.Errorf("Invalid destination %s", dest.Path)
log.Error(err)
return err
}
host := hostPort[0]
port := 80
if len(hostPort) == 2 {
var err error
port, err = strconv.Atoi(hostPort[1])
if err != nil {
err = fmt.Errorf("Invalid port %s", hostPort[1])
log.Error(err)
return err
}
}
return s.streamResolutionUsingCorrectResolverFor(host, port, stream)
}
// TODO: unimplemented
func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetProfileServer) error {
return nil
}
func (s *server) streamResolutionUsingCorrectResolverFor(host string, port int, stream pb.Destination_GetServer) error {
listener := newEndpointListener(stream, s.k8sAPI.GetOwnerKindAndName, s.enableTLS)
for _, resolver := range s.resolvers {
resolverCanResolve, err := resolver.canResolve(host, port)
if err != nil {
return fmt.Errorf("resolver [%+v] found error resolving host [%s] port[%d]: %v", resolver, host, port, err)
}
if resolverCanResolve {
return resolver.streamResolution(host, port, listener)
}
}
return fmt.Errorf("cannot find resolver for host [%s] port [%d]", host, port)
}
func buildResolversList(k8sDNSZone string, k8sAPI *k8s.API) ([]streamingDestinationResolver, error) {
var k8sDNSZoneLabels []string
if k8sDNSZone == "" {
k8sDNSZoneLabels = []string{}
} else {
var err error
k8sDNSZoneLabels, err = splitDNSName(k8sDNSZone)
if err != nil {
return nil, err
}
}
k8sResolver := newK8sResolver(k8sDNSZoneLabels, k8sAPI)
log.Infof("Adding k8s name resolver")
return []streamingDestinationResolver{k8sResolver}, nil
}