Skip to content

Commit 284b2e2

Browse files
ewoutpewoutp
authored
Merge pull request #74 from arangodb/watch-more-resources
Also watch changes in PVCs and Services
2 parents e0329ea + fba9d33 commit 284b2e2

File tree

6 files changed

+211
-115
lines changed

6 files changed

+211
-115
lines changed

‎pkg/deployment/deployment.go‎

Lines changed: 13 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -60,16 +60,12 @@ type deploymentEventType string
6060

6161
const (
6262
eventArangoDeploymentUpdated deploymentEventType = "ArangoDeploymentUpdated"
63-
eventPodAdded deploymentEventType = "PodAdded"
64-
eventPodUpdated deploymentEventType = "PodUpdated"
65-
eventPodDeleted deploymentEventType = "PodDeleted"
6663
)
6764

6865
// deploymentEvent holds an event passed from the controller to the deployment.
6966
type deploymentEvent struct {
7067
Type deploymentEventType
7168
Deployment *api.ArangoDeployment
72-
Pod *v1.Pod
7369
}
7470

7571
const (
@@ -91,6 +87,7 @@ type Deployment struct {
9187
eventsCli corev1.EventInterface
9288

9389
inspectTrigger trigger.Trigger
90+
updateDeploymentTrigger trigger.Trigger
9491
clientCache *clientCache
9592
recentInspectionErrors int
9693
clusterScalingIntegration *clusterScalingIntegration
@@ -121,7 +118,9 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
121118
}
122119

123120
go d.run()
124-
go d.listenForPodEvents()
121+
go d.listenForPodEvents(d.stopCh)
122+
go d.listenForPVCEvents(d.stopCh)
123+
go d.listenForServiceEvents(d.stopCh)
125124
if apiObject.Spec.GetMode() == api.DeploymentModeCluster {
126125
ci := newClusterScalingIntegration(d)
127126
d.clusterScalingIntegration = ci
@@ -217,20 +216,20 @@ func (d *Deployment) run() {
217216
// Got event from event queue
218217
switch event.Type {
219218
case eventArangoDeploymentUpdated:
220-
if err := d.handleArangoDeploymentUpdatedEvent(event); err != nil {
221-
d.failOnError(err, "Failed to handle deployment update")
222-
return
223-
}
224-
case eventPodAdded, eventPodUpdated, eventPodDeleted:
225-
// Pod event received, let's inspect soon
226-
d.inspectTrigger.Trigger()
219+
d.updateDeploymentTrigger.Trigger()
227220
default:
228221
panic("unknown event type" + event.Type)
229222
}
230223

231224
case <-d.inspectTrigger.Done():
232225
inspectionInterval = d.inspectDeployment(inspectionInterval)
233226

227+
case <-d.updateDeploymentTrigger.Done():
228+
if err := d.handleArangoDeploymentUpdatedEvent(); err != nil {
229+
d.failOnError(err, "Failed to handle deployment update")
230+
return
231+
}
232+
234233
case <-time.After(inspectionInterval):
235234
// Trigger inspection
236235
d.inspectTrigger.Trigger()
@@ -244,8 +243,8 @@ func (d *Deployment) run() {
244243
}
245244

246245
// handleArangoDeploymentUpdatedEvent is called when the deployment is updated by the user.
247-
func (d *Deployment) handleArangoDeploymentUpdatedEvent(event *deploymentEvent) error {
248-
log := d.deps.Log.With().Str("deployment", event.Deployment.GetName()).Logger()
246+
func (d *Deployment) handleArangoDeploymentUpdatedEvent() error {
247+
log := d.deps.Log.With().Str("deployment", d.apiObject.GetName()).Logger()
249248

250249
// Get the most recent version of the deployment from the API server
251250
current, err := d.deps.DatabaseCRCli.DatabaseV1alpha().ArangoDeployments(d.apiObject.GetNamespace()).Get(d.apiObject.GetName(), metav1.GetOptions{})

‎pkg/deployment/deployment_inspector.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,3 +101,8 @@ func (d *Deployment) inspectDeployment(lastInterval time.Duration) time.Duration
101101
}
102102
return nextInterval
103103
}
104+
105+
// triggerInspection ensures that an inspection is run soon.
106+
func (d *Deployment) triggerInspection() {
107+
d.inspectTrigger.Trigger()
108+
}

‎pkg/deployment/informers.go‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
//
2+
// DISCLAIMER
3+
//
4+
// Copyright 2018 ArangoDB GmbH, Cologne, Germany
5+
//
6+
// Licensed under the Apache License, Version 2.0 (the "License");
7+
// you may not use this file except in compliance with the License.
8+
// You may obtain a copy of the License at
9+
//
10+
// http://www.apache.org/licenses/LICENSE-2.0
11+
//
12+
// Unless required by applicable law or agreed to in writing, software
13+
// distributed under the License is distributed on an "AS IS" BASIS,
14+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
// See the License for the specific language governing permissions and
16+
// limitations under the License.
17+
//
18+
// Copyright holder is ArangoDB GmbH, Cologne, Germany
19+
//
20+
// Author Ewout Prangsma
21+
//
22+
23+
package deployment
24+
25+
import (
26+
"k8s.io/api/core/v1"
27+
"k8s.io/apimachinery/pkg/fields"
28+
"k8s.io/client-go/tools/cache"
29+
)
30+
31+
// listenForPodEvents keep listening for changes in pod until the given channel is closed.
32+
func (d *Deployment) listenForPodEvents(stopCh <-chan struct{}) {
33+
source := cache.NewListWatchFromClient(
34+
d.deps.KubeCli.CoreV1().RESTClient(),
35+
"pods",
36+
d.apiObject.GetNamespace(),
37+
fields.Everything())
38+
39+
getPod := func(obj interface{}) (*v1.Pod, bool) {
40+
pod, ok := obj.(*v1.Pod)
41+
if !ok {
42+
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
43+
if !ok {
44+
return nil, false
45+
}
46+
pod, ok = tombstone.Obj.(*v1.Pod)
47+
return pod, ok
48+
}
49+
return pod, true
50+
}
51+
52+
_, informer := cache.NewIndexerInformer(source, &v1.Pod{}, 0, cache.ResourceEventHandlerFuncs{
53+
AddFunc: func(obj interface{}) {
54+
if p, ok := getPod(obj); ok && d.isOwnerOf(p) {
55+
d.triggerInspection()
56+
}
57+
},
58+
UpdateFunc: func(oldObj, newObj interface{}) {
59+
if p, ok := getPod(newObj); ok && d.isOwnerOf(p) {
60+
d.triggerInspection()
61+
}
62+
},
63+
DeleteFunc: func(obj interface{}) {
64+
if p, ok := getPod(obj); ok && d.isOwnerOf(p) {
65+
d.triggerInspection()
66+
}
67+
},
68+
}, cache.Indexers{})
69+
70+
informer.Run(stopCh)
71+
}
72+
73+
// listenForPVCEvents keep listening for changes in PVC's until the given channel is closed.
74+
func (d *Deployment) listenForPVCEvents(stopCh <-chan struct{}) {
75+
source := cache.NewListWatchFromClient(
76+
d.deps.KubeCli.CoreV1().RESTClient(),
77+
"persistentvolumeclaims",
78+
d.apiObject.GetNamespace(),
79+
fields.Everything())
80+
81+
getPVC := func(obj interface{}) (*v1.PersistentVolumeClaim, bool) {
82+
pvc, ok := obj.(*v1.PersistentVolumeClaim)
83+
if !ok {
84+
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
85+
if !ok {
86+
return nil, false
87+
}
88+
pvc, ok = tombstone.Obj.(*v1.PersistentVolumeClaim)
89+
return pvc, ok
90+
}
91+
return pvc, true
92+
}
93+
94+
_, informer := cache.NewIndexerInformer(source, &v1.PersistentVolumeClaim{}, 0, cache.ResourceEventHandlerFuncs{
95+
AddFunc: func(obj interface{}) {
96+
if p, ok := getPVC(obj); ok && d.isOwnerOf(p) {
97+
d.triggerInspection()
98+
}
99+
},
100+
UpdateFunc: func(oldObj, newObj interface{}) {
101+
if p, ok := getPVC(newObj); ok && d.isOwnerOf(p) {
102+
d.triggerInspection()
103+
}
104+
},
105+
DeleteFunc: func(obj interface{}) {
106+
if p, ok := getPVC(obj); ok && d.isOwnerOf(p) {
107+
d.triggerInspection()
108+
}
109+
},
110+
}, cache.Indexers{})
111+
112+
informer.Run(stopCh)
113+
}
114+
115+
// listenForServiceEvents keep listening for changes in Service's until the given channel is closed.
116+
func (d *Deployment) listenForServiceEvents(stopCh <-chan struct{}) {
117+
source := cache.NewListWatchFromClient(
118+
d.deps.KubeCli.CoreV1().RESTClient(),
119+
"services",
120+
d.apiObject.GetNamespace(),
121+
fields.Everything())
122+
123+
getService := func(obj interface{}) (*v1.Service, bool) {
124+
service, ok := obj.(*v1.Service)
125+
if !ok {
126+
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
127+
if !ok {
128+
return nil, false
129+
}
130+
service, ok = tombstone.Obj.(*v1.Service)
131+
return service, ok
132+
}
133+
return service, true
134+
}
135+
136+
_, informer := cache.NewIndexerInformer(source, &v1.Service{}, 0, cache.ResourceEventHandlerFuncs{
137+
AddFunc: func(obj interface{}) {
138+
if s, ok := getService(obj); ok && d.isOwnerOf(s) {
139+
d.triggerInspection()
140+
}
141+
},
142+
UpdateFunc: func(oldObj, newObj interface{}) {
143+
if s, ok := getService(newObj); ok && d.isOwnerOf(s) {
144+
d.triggerInspection()
145+
}
146+
},
147+
DeleteFunc: func(obj interface{}) {
148+
if s, ok := getService(obj); ok && d.isOwnerOf(s) {
149+
d.triggerInspection()
150+
}
151+
},
152+
}, cache.Indexers{})
153+
154+
informer.Run(stopCh)
155+
}

‎pkg/deployment/pod_informer.go‎

Lines changed: 0 additions & 80 deletions
This file was deleted.

‎pkg/deployment/resources/services.go‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -34,18 +34,23 @@ func (r *Resources) EnsureServices() error {
3434
owner := apiObject.AsOwner()
3535
spec := r.context.GetSpec()
3636

37-
log.Debug().Msg("creating services...")
38-
39-
if _, err := k8sutil.CreateHeadlessService(kubecli, apiObject, owner); err != nil {
37+
svcName, newlyCreated, err := k8sutil.CreateHeadlessService(kubecli, apiObject, owner)
38+
if err != nil {
4039
log.Debug().Err(err).Msg("Failed to create headless service")
4140
return maskAny(err)
4241
}
42+
if newlyCreated {
43+
log.Debug().Str("service", svcName).Msg("Created headless service")
44+
}
4345
single := spec.GetMode().HasSingleServers()
44-
svcName, err := k8sutil.CreateDatabaseClientService(kubecli, apiObject, single, owner)
46+
svcName, newlyCreated, err = k8sutil.CreateDatabaseClientService(kubecli, apiObject, single, owner)
4547
if err != nil {
4648
log.Debug().Err(err).Msg("Failed to create database client service")
4749
return maskAny(err)
4850
}
51+
if newlyCreated {
52+
log.Debug().Str("service", svcName).Msg("Created database client service")
53+
}
4954
status := r.context.GetStatus()
5055
if status.ServiceName != svcName {
5156
status.ServiceName = svcName
@@ -55,11 +60,14 @@ func (r *Resources) EnsureServices() error {
5560
}
5661

5762
if spec.Sync.IsEnabled() {
58-
svcName, err := k8sutil.CreateSyncMasterClientService(kubecli, apiObject, owner)
63+
svcName, newlyCreated, err := k8sutil.CreateSyncMasterClientService(kubecli, apiObject, owner)
5964
if err != nil {
6065
log.Debug().Err(err).Msg("Failed to create syncmaster client service")
6166
return maskAny(err)
6267
}
68+
if newlyCreated {
69+
log.Debug().Str("service", svcName).Msg("Created syncmasters service")
70+
}
6371
status := r.context.GetStatus()
6472
if status.SyncServiceName != svcName {
6573
status.SyncServiceName = svcName

0 commit comments

Comments
 (0)