forked from openshift/origin
/
plugin.go
178 lines (150 loc) · 5.57 KB
/
plugin.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
package templaterouter
import (
"fmt"
"strconv"
"text/template"
kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
ktypes "github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
routeapi "github.com/openshift/origin/pkg/route/api"
)
// TemplatePlugin implements the router.Plugin interface to provide
// a template based, backend-agnostic router.
type TemplatePlugin struct {
Router router
}
type TemplatePluginConfig struct {
TemplatePath string
ReloadScriptPath string
DefaultCertificate string
StatsPort int
StatsUsername string
StatsPassword string
PeerService ktypes.NamespacedName
}
// router controls the interaction of the plugin with the underlying router implementation
type router interface {
// Mutative operations in this interface do not return errors.
// The only error state for these methods is when an unknown
// frontend key is used; all call sites make certain the frontend
// is created.
// CreateServiceUnit creates a new service named with the given id.
CreateServiceUnit(id string)
// FindServiceUnit finds the service with the given id.
FindServiceUnit(id string) (v ServiceUnit, ok bool)
// AddEndpoints adds new Endpoints for the given id.Returns true if a change was made
// and the state should be stored with Commit().
AddEndpoints(id string, endpoints []Endpoint) bool
// DeleteEndpoints deletes the endpoints for the frontend with the given id.
DeleteEndpoints(id string)
// AddRoute adds a route for the given id. Returns true if a change was made
// and the state should be stored with Commit().
AddRoute(id string, route *routeapi.Route) bool
// RemoveRoute removes the given route for the given id.
RemoveRoute(id string, route *routeapi.Route)
// Commit refreshes the backend and persists the router state.
Commit() error
}
// NewTemplatePlugin creates a new TemplatePlugin.
func NewTemplatePlugin(cfg TemplatePluginConfig) (*TemplatePlugin, error) {
masterTemplate := template.Must(template.New("config").ParseFiles(cfg.TemplatePath))
templates := map[string]*template.Template{}
for _, template := range masterTemplate.Templates() {
if template == masterTemplate {
continue
}
templates[template.Name()] = template
}
templateRouterCfg := templateRouterCfg{
templates: templates,
reloadScriptPath: cfg.ReloadScriptPath,
defaultCertificate: cfg.DefaultCertificate,
statsUser: cfg.StatsUsername,
statsPassword: cfg.StatsPassword,
statsPort: cfg.StatsPort,
peerEndpointsKey: peerEndpointsKey(cfg.PeerService),
}
router, err := newTemplateRouter(templateRouterCfg)
return &TemplatePlugin{router}, err
}
// HandleEndpoints processes watch events on the Endpoints resource.
func (p *TemplatePlugin) HandleEndpoints(eventType watch.EventType, endpoints *kapi.Endpoints) error {
key := endpointsKey(*endpoints)
glog.V(4).Infof("Processing %d Endpoints for Name: %v (%v)", len(endpoints.Subsets), endpoints.Name, eventType)
for i, s := range endpoints.Subsets {
glog.V(4).Infof(" Subset %d : %#v", i, s)
}
if _, ok := p.Router.FindServiceUnit(key); !ok {
p.Router.CreateServiceUnit(key)
}
switch eventType {
case watch.Added, watch.Modified:
glog.V(4).Infof("Modifying endpoints for %s", key)
routerEndpoints := createRouterEndpoints(endpoints)
key := endpointsKey(*endpoints)
commit := p.Router.AddEndpoints(key, routerEndpoints)
if commit {
return p.Router.Commit()
}
}
return nil
}
// HandleRoute processes watch events on the Route resource.
func (p *TemplatePlugin) HandleRoute(eventType watch.EventType, route *routeapi.Route) error {
key := routeKey(*route)
if _, ok := p.Router.FindServiceUnit(key); !ok {
glog.V(4).Infof("Creating new frontend for key: %v", key)
p.Router.CreateServiceUnit(key)
}
switch eventType {
case watch.Added, watch.Modified:
glog.V(4).Infof("Modifying routes for %s", key)
commit := p.Router.AddRoute(key, route)
if commit {
return p.Router.Commit()
}
case watch.Deleted:
glog.V(4).Infof("Deleting routes for %s", key)
p.Router.RemoveRoute(key, route)
return p.Router.Commit()
}
return nil
}
// routeKey returns the internal router key to use for the given Route.
func routeKey(route routeapi.Route) string {
return fmt.Sprintf("%s/%s", route.Namespace, route.ServiceName)
}
// endpointsKey returns the internal router key to use for the given Endpoints.
func endpointsKey(endpoints kapi.Endpoints) string {
return fmt.Sprintf("%s/%s", endpoints.Namespace, endpoints.Name)
}
// peerServiceKey may be used by the underlying router when handling endpoints to identify
// endpoints that belong to its peers. THIS MUST FOLLOW THE KEY STRATEGY OF endpointsKey. It
// receives a NamespacedName that is created from the service that is added by the oadm command
func peerEndpointsKey(namespacedName ktypes.NamespacedName) string {
return fmt.Sprintf("%s/%s", namespacedName.Namespace, namespacedName.Name)
}
// createRouterEndpoints creates openshift router endpoints based on k8s endpoints
func createRouterEndpoints(endpoints *kapi.Endpoints) []Endpoint {
out := make([]Endpoint, 0, len(endpoints.Subsets)*4)
// TODO: review me for sanity
for _, s := range endpoints.Subsets {
for _, a := range s.Addresses {
for _, p := range s.Ports {
ep := Endpoint{
ID: fmt.Sprintf("%s:%d", a.IP, p.Port),
IP: a.IP,
Port: strconv.Itoa(p.Port),
}
if a.TargetRef != nil {
ep.TargetName = a.TargetRef.Name
} else {
ep.TargetName = ep.IP
}
out = append(out, ep)
}
}
}
return out
}