/
check_migrations.go
81 lines (67 loc) · 2.47 KB
/
check_migrations.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
package controller
import (
"context"
"golang.org/x/exp/slices"
corev1 "k8s.io/api/core/v1"
"k8s.io/client-go/tools/record"
"github.com/authzed/controller-idioms/handler"
"github.com/authzed/controller-idioms/hash"
"github.com/authzed/spicedb-operator/pkg/apis/authzed/v1alpha1"
"github.com/authzed/spicedb-operator/pkg/metadata"
)
const (
EventRunningMigrations = "RunningMigrations"
HandlerDeploymentKey handler.Key = "deploymentChain"
HandlerMigrationRunKey handler.Key = "runMigration"
HandlerWaitForMigrationsKey handler.Key = "waitForMigrationChain"
)
type MigrationCheckHandler struct {
recorder record.EventRecorder
nextMigrationRunHandler handler.ContextHandler
nextWaitForJobHandler handler.ContextHandler
nextDeploymentHandler handler.ContextHandler
}
func (m *MigrationCheckHandler) Handle(ctx context.Context) {
migrationHash := CtxMigrationHash.MustValue(ctx)
hasJob := false
hasDeployment := false
for _, d := range CtxDeployments.MustValue(ctx) {
if d.Annotations != nil && hash.SecureEqual(d.Annotations[metadata.SpiceDBMigrationRequirementsKey], migrationHash) {
hasDeployment = true
break
}
}
for _, j := range CtxJobs.MustValue(ctx) {
if j.Annotations != nil && hash.SecureEqual(j.Annotations[metadata.SpiceDBMigrationRequirementsKey], migrationHash) {
hasJob = true
ctx = CtxCurrentMigrationJob.WithValue(ctx, j)
break
}
}
// don't handle migrations at all if `skipMigrations` is set, if the
// `memory` datastore is used, or if the update graph says there are no
// migrations for this step.
config := CtxConfig.MustValue(ctx)
if config.SkipMigrations || config.DatastoreEngine == "memory" {
m.nextDeploymentHandler.Handle(ctx)
return
}
status := CtxCluster.MustValue(ctx).Status
if status.CurrentVersion != nil && !slices.Contains(status.CurrentVersion.Attributes, v1alpha1.SpiceDBVersionAttributesMigration) {
m.nextDeploymentHandler.Handle(ctx)
return
}
// if there's no job and no (updated) deployment, create the job
if !hasDeployment && !hasJob {
m.recorder.Eventf(CtxCluster.MustValue(ctx), corev1.EventTypeNormal, EventRunningMigrations, "Running migration job for %s", CtxConfig.MustValue(ctx).TargetSpiceDBImage)
m.nextMigrationRunHandler.Handle(ctx)
return
}
// if there's a job but no (updated) deployment, wait for the job
if hasJob && !hasDeployment {
m.nextWaitForJobHandler.Handle(ctx)
return
}
// if the deployment is up to date, continue
m.nextDeploymentHandler.Handle(ctx)
}