forked from hashicorp/consul
-
Notifications
You must be signed in to change notification settings - Fork 0
/
controller.go
148 lines (128 loc) · 4.14 KB
/
controller.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
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package reaper
import (
"context"
"time"
"github.com/hernad/consul/internal/controller"
"github.com/hernad/consul/internal/resource"
"github.com/hernad/consul/proto-public/pbresource"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
const (
statusKeyReaperController = "consul.io/reaper-controller"
secondPassDelay = 30 * time.Second
conditionTypeFirstPassCompleted = "FirstPassCompleted"
)
// RegisterControllers registers controllers for the tombstone type.
func RegisterControllers(mgr *controller.Manager) {
mgr.Register(reaperController())
}
func reaperController() controller.Controller {
return controller.ForType(resource.TypeV1Tombstone).
WithReconciler(newReconciler())
}
func newReconciler() *tombstoneReconciler {
return &tombstoneReconciler{
timeNow: time.Now,
}
}
type tombstoneReconciler struct {
// Testing shim
timeNow func() time.Time
}
// Deletes all owned (child) resources of an owner (parent) resource.
//
// The reconciliation for tombstones is split into two passes.
// The first pass attempts to delete child resources created before the owner resource was deleted.
// The second pass is run after a reasonable delay to delete child resources that may have been
// created during or after the completion of the first pass.
func (r *tombstoneReconciler) Reconcile(ctx context.Context, rt controller.Runtime, req controller.Request) error {
rsp, err := rt.Client.Read(ctx, &pbresource.ReadRequest{Id: req.ID})
switch {
case status.Code(err) == codes.NotFound:
// tombstone not found. nothing to do
return nil
case err != nil:
// retry later
return err
}
res := rsp.Resource
var tombstone pbresource.Tombstone
if err := res.Data.UnmarshalTo(&tombstone); err != nil {
return err
}
firstPassCompletedOnEntry := isFirstPassCompleted(res)
// Corner case:
// Check secondPassDelay has elasped since first pass in cases where queued
// reconciliation requests are lost between the first and second pass
// (e.g. controller relocated to a different server due to raft leadership
// change).
if firstPassCompletedOnEntry && !r.secondPassDelayElapsed(res.Status[statusKeyReaperController]) {
return controller.RequeueAfter(secondPassDelay)
}
// Retrieve owner's children
listRsp, err := rt.Client.ListByOwner(ctx, &pbresource.ListByOwnerRequest{Owner: tombstone.Owner})
if err != nil {
return err
}
// Attempt to delete each child
for _, child := range listRsp.Resources {
_, err := rt.Client.Delete(ctx, &pbresource.DeleteRequest{Id: child.Id})
if err != nil {
return err
}
}
if firstPassCompletedOnEntry {
// we just did the second pass -> delete tombstone
_, err := rt.Client.Delete(ctx, &pbresource.DeleteRequest{Id: res.Id})
if err != nil {
// tombstone deletion failed, just retry
return err
}
// tombstone delete succeeded and reconciliation complete
return nil
} else {
// we just did the first pass -> queue up the second pass
_, err = rt.Client.WriteStatus(ctx, &pbresource.WriteStatusRequest{
Id: res.Id,
Key: statusKeyReaperController,
Status: &pbresource.Status{
ObservedGeneration: res.Generation,
Conditions: []*pbresource.Condition{
{
Type: conditionTypeFirstPassCompleted,
State: pbresource.Condition_STATE_TRUE,
Reason: "Success",
Message: "First pass of child resource deletion completed",
},
},
},
})
if err != nil {
return err
}
return controller.RequeueAfter(secondPassDelay)
}
}
func (r *tombstoneReconciler) secondPassDelayElapsed(status *pbresource.Status) bool {
firstPassTime := status.UpdatedAt.AsTime()
return firstPassTime.Add(secondPassDelay).Before(r.timeNow())
}
func isFirstPassCompleted(res *pbresource.Resource) bool {
if res.Status == nil {
return false
}
status, ok := res.Status[statusKeyReaperController]
if !ok {
return false
}
// First time through, first and second pass ahead of us
if len(status.Conditions) == 0 {
return false
}
// Single condition "FirstPassCompleted"
condition := status.Conditions[0]
return condition.State == pbresource.Condition_STATE_TRUE
}