-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
handler.go
187 lines (161 loc) · 5.05 KB
/
handler.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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
// Copyright 2020, OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package k8sobserver
import (
"fmt"
"reflect"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/tools/cache"
"github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer"
)
// handler handles k8s cache informer callbacks.
type handler struct {
// idNamespace should be some unique token to distinguish multiple handler instances.
idNamespace string
// watcher is the callback for discovered endpoints.
watcher observer.Notify
}
// OnAdd is called in response to a pod being added.
func (h *handler) OnAdd(obj interface{}) {
pod, ok := obj.(*v1.Pod)
if !ok {
return
}
h.watcher.OnAdd(h.convertPodToEndpoints(pod))
}
// convertPodToEndpoints converts a pod instance into a slice of endpoints. The endpoints
// include the pod itself as well as an endpoint for each container port that is mapped
// to a container that is in a running state.
func (h *handler) convertPodToEndpoints(pod *v1.Pod) []observer.Endpoint {
podID := observer.EndpointID(fmt.Sprintf("%s/%s", h.idNamespace, pod.UID))
podIP := pod.Status.PodIP
podDetails := observer.Pod{
Annotations: pod.Annotations,
Labels: pod.Labels,
Name: pod.Name,
}
endpoints := []observer.Endpoint{{
ID: podID,
Target: podIP,
Details: podDetails,
}}
// Map of running containers by name.
containerRunning := map[string]bool{}
for _, container := range pod.Status.ContainerStatuses {
if container.State.Running != nil {
containerRunning[container.Name] = true
}
}
// Create endpoint for each named container port.
for _, container := range pod.Spec.Containers {
if !containerRunning[container.Name] {
continue
}
for _, port := range container.Ports {
endpointID := observer.EndpointID(
fmt.Sprintf(
"%s/%s(%d)", podID, port.Name, port.ContainerPort,
),
)
endpoints = append(endpoints, observer.Endpoint{
ID: endpointID,
Target: fmt.Sprintf("%s:%d", podIP, port.ContainerPort),
Details: observer.Port{
Pod: podDetails,
Name: port.Name,
Port: uint16(port.ContainerPort),
Transport: getTransport(port.Protocol),
},
})
}
}
return endpoints
}
func getTransport(protocol v1.Protocol) observer.Transport {
switch protocol {
case v1.ProtocolTCP:
return observer.ProtocolTCP
case v1.ProtocolUDP:
return observer.ProtocolUDP
}
return observer.ProtocolUnknown
}
// OnUpdate is called in response to an existing pod changing.
func (h *handler) OnUpdate(oldObj, newObj interface{}) {
oldPod, ok := oldObj.(*v1.Pod)
if !ok {
return
}
newPod, ok := newObj.(*v1.Pod)
if !ok {
return
}
oldEndpoints := map[observer.EndpointID]observer.Endpoint{}
newEndpoints := map[observer.EndpointID]observer.Endpoint{}
// Convert pods to endpoints and map by ID for easier lookup.
for _, e := range h.convertPodToEndpoints(oldPod) {
oldEndpoints[e.ID] = e
}
for _, e := range h.convertPodToEndpoints(newPod) {
newEndpoints[e.ID] = e
}
var removedEndpoints, updatedEndpoints, addedEndpoints []observer.Endpoint
// Find endpoints that are present in oldPod and newPod and see if they've
// changed. Otherwise if it wasn't in oldPod it's a new endpoint.
for _, e := range newEndpoints {
if existing, ok := oldEndpoints[e.ID]; ok {
if !reflect.DeepEqual(existing, e) {
updatedEndpoints = append(updatedEndpoints, e)
}
} else {
addedEndpoints = append(addedEndpoints, e)
}
}
// If an endpoint is present in the oldPod but not in the newPod then
// send as removed.
for _, e := range oldEndpoints {
if _, ok := newEndpoints[e.ID]; !ok {
removedEndpoints = append(removedEndpoints, e)
}
}
if len(removedEndpoints) > 0 {
h.watcher.OnRemove(removedEndpoints)
}
if len(updatedEndpoints) > 0 {
h.watcher.OnChange(updatedEndpoints)
}
if len(addedEndpoints) > 0 {
h.watcher.OnAdd(addedEndpoints)
}
// TODO: can changes be missed where a pod is deleted but we don't
// send remove notifications for some of its endpoints? If not provable
// then maybe keep track of pod -> endpoint association to be sure
// they are all cleaned up.
}
// OnDelete is called in response to a pod being deleted.
func (h *handler) OnDelete(obj interface{}) {
var pod *v1.Pod
switch o := obj.(type) {
case *cache.DeletedFinalStateUnknown:
// Assuming we never saw the pod state where new endpoints would have been created
// to begin with it seems that we can't leak endpoints here.
pod = o.Obj.(*v1.Pod)
case *v1.Pod:
pod = o
}
if pod == nil {
return
}
h.watcher.OnRemove(h.convertPodToEndpoints(pod))
}