forked from rancher/rancher
-
Notifications
You must be signed in to change notification settings - Fork 1
/
workload_endpoints.go
145 lines (134 loc) · 3.94 KB
/
workload_endpoints.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
package endpoints
import (
"fmt"
"strings"
workloadutil "github.com/rancher/rancher/pkg/controllers/user/workload"
"github.com/rancher/types/apis/core/v1"
managementv3 "github.com/rancher/types/apis/management.cattle.io/v3"
"github.com/rancher/types/apis/project.cattle.io/v3"
"github.com/sirupsen/logrus"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/labels"
)
// This controller is responsible for monitoring workloads
// and setting public endpoints on them based on HostPort pods
// and NodePort services backing up the workload
type WorkloadEndpointsController struct {
serviceLister v1.ServiceLister
podLister v1.PodLister
WorkloadController workloadutil.CommonController
machinesLister managementv3.NodeLister
clusterName string
}
func (c *WorkloadEndpointsController) UpdateEndpoints(key string, obj *workloadutil.Workload) error {
if obj == nil && key != workloadutil.AllWorkloads {
return nil
}
// do not update endpoints for job, cronJob and for workload owned by controller (ReplicaSet)
if strings.EqualFold(obj.Kind, "job") || strings.EqualFold(obj.Kind, "cronJob") {
return nil
}
for _, o := range obj.OwnerReferences {
if *o.Controller {
return nil
}
}
var workloads []*workloadutil.Workload
var services []*corev1.Service
var err error
if strings.HasSuffix(key, workloadutil.AllWorkloads) {
namespace := ""
if !strings.EqualFold(key, allEndpoints) {
namespace = strings.TrimSuffix(key, fmt.Sprintf("/%s", allEndpoints))
}
workloads, err = c.WorkloadController.GetAllWorkloads(namespace)
if err != nil {
return err
}
services, err = c.serviceLister.List(namespace, labels.NewSelector())
} else {
services, err = c.serviceLister.List(obj.Namespace, labels.NewSelector())
workloads = append(workloads, obj)
}
if err != nil {
return err
}
nodeNameToMachine, err := getNodeNameToMachine(c.clusterName, c.machinesLister)
if err != nil {
return err
}
for _, w := range workloads {
// 1. Get endpoints from services
var newPublicEps []v3.PublicEndpoint
for _, svc := range services {
set := labels.Set{}
for key, val := range svc.Spec.Selector {
set[key] = val
}
selector := labels.SelectorFromSet(set)
found := false
if selector.Matches(labels.Set(w.Labels)) {
// direct selector match
found = true
} else {
// match based off the workload
value, ok := svc.Annotations[workloadutil.WorkloadAnnotation]
if !ok || value == "" {
continue
}
workloadIDs := strings.Split(value, ",")
for _, workloadID := range workloadIDs {
splitted := strings.Split(workloadID, ":")
if len(splitted) != 3 {
continue
}
namespace := splitted[1]
name := splitted[2]
if w.Name == name && w.Namespace == namespace {
found = true
break
}
}
}
if found {
eps, err := convertServiceToPublicEndpoints(svc, "", nil)
if err != nil {
return err
}
newPublicEps = append(newPublicEps, eps...)
}
}
// 2. Get endpoints from HostPort pods matching the selector
set := labels.Set{}
for key, val := range w.SelectorLabels {
set[key] = val
}
pods, err := c.podLister.List(w.Namespace, labels.SelectorFromSet(set))
if err != nil {
return err
}
for _, pod := range pods {
eps, err := convertHostPortToEndpoint(pod, c.clusterName, nodeNameToMachine[pod.Spec.NodeName])
if err != nil {
return err
}
newPublicEps = append(newPublicEps, eps...)
}
existingPublicEps := getPublicEndpointsFromAnnotations(w.Annotations)
if areEqualEndpoints(existingPublicEps, newPublicEps) {
return nil
}
epsToUpdate, err := publicEndpointsToString(newPublicEps)
if err != nil {
return err
}
logrus.Infof("Updating workload [%s] with public endpoints [%v]", key, epsToUpdate)
annotations := map[string]string{
endpointsAnnotation: epsToUpdate,
}
if err = c.WorkloadController.UpdateWorkload(w, annotations); err != nil {
return err
}
}
return nil
}