Skip to content

Commit 2fe9735

Browse files
carlydfcarlydf
committed
catch controller panic in tests, fix bug, callback stopfunc correctly
1 parent 858f5a9 commit 2fe9735

6 files changed

Lines changed: 42 additions & 25 deletions

File tree

‎Makefile‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,10 @@ ALL_TEST_TAGS = test_dep
99
##### Variables ######
1010

1111
ROOT := $(shell git rev-parse --show-toplevel)
12-
LOCALBIN := .bin
12+
## Location to install dependencies to
13+
LOCALBIN ?= $(shell pwd)/bin
14+
$(LOCALBIN):
15+
mkdir -p $(LOCALBIN)
1316
STAMPDIR := .stamp
1417
export PATH := $(ROOT)/$(LOCALBIN):$(PATH)
1518
GOINSTALL := GOBIN=$(ROOT)/$(LOCALBIN) go install
@@ -109,7 +112,6 @@ set -e; \
109112
package=$(2)@$(3) ;\
110113
printf $(COLOR) "Downloading $${package}" ;\
111114
tmpdir=$$(mktemp -d) ;\
112-
GOBIN=$${tmpdir} go install $${package} ;\
113115
mv $${tmpdir}/$$(basename "$$(echo "$(1)" | sed "s/-$(3)$$//")") $(1) ;\
114116
rm -rf $${tmpdir} ;\
115117
}

‎internal/controller/worker_controller.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,9 @@ type TemporalWorkerDeploymentReconciler struct {
3737
client.Client
3838
Scheme *runtime.Scheme
3939
TemporalClientPool *clientpool.ClientPool
40+
41+
// Disables panic recovery if true
42+
DisableRecoverPanic bool
4043
}
4144

4245
//+kubebuilder:rbac:groups=temporal.io,resources=temporalworkerdeployments,verbs=get;list;watch;create;update;patch;delete
@@ -191,11 +194,13 @@ func (r *TemporalWorkerDeploymentReconciler) SetupWithManager(mgr ctrl.Manager)
191194
return err
192195
}
193196

197+
recoverPanic := !r.DisableRecoverPanic
194198
return ctrl.NewControllerManagedBy(mgr).
195199
For(&temporaliov1alpha1.TemporalWorkerDeployment{}).
196200
Owns(&appsv1.Deployment{}).
197201
WithOptions(controller.Options{
198202
MaxConcurrentReconciles: 100,
203+
RecoverPanic: &recoverPanic,
199204
}).
200205
Complete(r)
201206
}

‎internal/planner/planner.go‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -381,13 +381,15 @@ func handleProgressiveRollout(
381381
}
382382

383383
// Move to the next step if it has been long enough since the last update
384-
if rampLastModifiedAt.Add(currentStep.PauseDuration.Duration).Before(currentTime) {
385-
if i < len(steps)-1 {
386-
vcfg.RampPercentage = steps[i+1].RampPercentage
387-
return vcfg
388-
} else {
389-
vcfg.SetCurrent = true
390-
return vcfg
384+
if rampLastModifiedAt != nil {
385+
if rampLastModifiedAt.Add(currentStep.PauseDuration.Duration).Before(currentTime) {
386+
if i < len(steps)-1 {
387+
vcfg.RampPercentage = steps[i+1].RampPercentage
388+
return vcfg
389+
} else {
390+
vcfg.SetCurrent = true
391+
return vcfg
392+
}
391393
}
392394
}
393395

‎internal/tests/internal/env_helpers.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -122,9 +122,10 @@ func setupTestEnvironment(t *testing.T) (*rest.Config, client.Client, manager.Ma
122122

123123
// Set up controller
124124
reconciler := &controller.TemporalWorkerDeploymentReconciler{
125-
Client: mgr.GetClient(),
126-
Scheme: mgr.GetScheme(),
127-
TemporalClientPool: clientPool,
125+
Client: mgr.GetClient(),
126+
Scheme: mgr.GetScheme(),
127+
TemporalClientPool: clientPool,
128+
DisableRecoverPanic: true,
128129
}
129130
err = reconciler.SetupWithManager(mgr)
130131
if err != nil {

‎internal/tests/internal/integration_test.go‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -69,9 +69,9 @@ func TestIntegration(t *testing.T) {
6969

7070
tests := map[string]testCase{
7171
"all-at-once-rollout-2-replicas": {
72-
input: testhelpers.ModifyObj(common.MakeTWDWithName("all-at-once-rollout-2-replicas"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
72+
input: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("all-at-once-rollout-2-replicas"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
7373
obj.Spec.RolloutStrategy.Strategy = temporaliov1alpha1.UpdateAllAtOnce
74-
obj.Spec.Template = common.MakeHelloWorldPodSpec("v1")
74+
obj.Spec.Template = testhelpers.MakeHelloWorldPodSpec("v1")
7575
replicas := int32(2)
7676
obj.Spec.Replicas = &replicas
7777
obj.Spec.WorkerOptions = temporaliov1alpha1.WorkerOptions{
@@ -90,12 +90,12 @@ func TestIntegration(t *testing.T) {
9090
TargetVersion: nil,
9191
CurrentVersion: &temporaliov1alpha1.CurrentWorkerDeploymentVersion{
9292
BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
93-
VersionID: common.MakeVersionId(testNamespace.Name, "all-at-once-rollout-2-replicas", "v1"),
93+
VersionID: testhelpers.MakeVersionId(testNamespace.Name, "all-at-once-rollout-2-replicas", "v1"),
9494
Deployment: &corev1.ObjectReference{
9595
Namespace: testNamespace.Name,
9696
Name: k8s.ComputeVersionedDeploymentName(
9797
"all-at-once-rollout-2-replicas",
98-
common.MakeBuildId("all-at-once-rollout-2-replicas", "v1", nil),
98+
testhelpers.MakeBuildId("all-at-once-rollout-2-replicas", "v1", nil),
9999
),
100100
},
101101
},
@@ -107,12 +107,12 @@ func TestIntegration(t *testing.T) {
107107
},
108108
},
109109
"progressive-rollout-expect-first-step": {
110-
input: testhelpers.ModifyObj(common.MakeTWDWithName("progressive-rollout-expect-first-step"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
110+
input: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("progressive-rollout-expect-first-step"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
111111
obj.Spec.RolloutStrategy.Strategy = temporaliov1alpha1.UpdateProgressive
112112
obj.Spec.RolloutStrategy.Steps = []temporaliov1alpha1.RolloutStep{
113-
{5, metav1.Duration{time.Hour}},
113+
{RampPercentage: 5, PauseDuration: metav1.Duration{Duration: time.Hour}},
114114
}
115-
obj.Spec.Template = common.MakeHelloWorldPodSpec("v1")
115+
obj.Spec.Template = testhelpers.MakeHelloWorldPodSpec("v1")
116116
obj.Spec.WorkerOptions = temporaliov1alpha1.WorkerOptions{
117117
TemporalConnection: "progressive-rollout-expect-first-step",
118118
TemporalNamespace: ts.GetDefaultNamespace(),
@@ -128,12 +128,12 @@ func TestIntegration(t *testing.T) {
128128
expectedStatus: &temporaliov1alpha1.TemporalWorkerDeploymentStatus{
129129
TargetVersion: &temporaliov1alpha1.TargetWorkerDeploymentVersion{
130130
BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
131-
VersionID: common.MakeVersionId(testNamespace.Name, "progressive-rollout-expect-first-step", "v1"),
131+
VersionID: testhelpers.MakeVersionId(testNamespace.Name, "progressive-rollout-expect-first-step", "v1"),
132132
Deployment: &corev1.ObjectReference{
133133
Namespace: testNamespace.Name,
134134
Name: k8s.ComputeVersionedDeploymentName(
135135
"progressive-rollout-expect-first-step",
136-
common.MakeBuildId("progressive-rollout-expect-first-step", "v1", nil),
136+
testhelpers.MakeBuildId("progressive-rollout-expect-first-step", "v1", nil),
137137
),
138138
},
139139
},
@@ -145,12 +145,12 @@ func TestIntegration(t *testing.T) {
145145
CurrentVersion: nil,
146146
RampingVersion: &temporaliov1alpha1.TargetWorkerDeploymentVersion{
147147
BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
148-
VersionID: common.MakeVersionId(testNamespace.Name, "progressive-rollout-expect-first-step", "v1"),
148+
VersionID: testhelpers.MakeVersionId(testNamespace.Name, "progressive-rollout-expect-first-step", "v1"),
149149
Deployment: &corev1.ObjectReference{
150150
Namespace: testNamespace.Name,
151151
Name: k8s.ComputeVersionedDeploymentName(
152152
"progressive-rollout-expect-first-step",
153-
common.MakeBuildId("progressive-rollout-expect-first-step", "v1", nil),
153+
testhelpers.MakeBuildId("progressive-rollout-expect-first-step", "v1", nil),
154154
),
155155
},
156156
},

‎internal/tests/internal/workers.go‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ func newClient(ctx context.Context, hostPort, namespace string) (client.Client,
9292
return c, nil
9393
}
9494

95+
// callback is a function that can be called multiple times.
9596
func runHelloWorldWorker(ctx context.Context, podTemplateSpec corev1.PodTemplateSpec, callback func(stopFunc func(), err error)) {
9697
w, stopFunc, err := newVersionedWorker(ctx, podTemplateSpec)
9798
defer func() {
@@ -108,7 +109,6 @@ func runHelloWorldWorker(ctx context.Context, podTemplateSpec corev1.PodTemplate
108109
sleep := func(ctx context.Context, seconds uint) error {
109110
time.Sleep(time.Duration(seconds) * time.Second)
110111
return nil
111-
//return temporal.NewNonRetryableApplicationError("oops", "", nil)
112112
}
113113

114114
helloWorld := func(ctx workflow.Context) (string, error) {
@@ -134,7 +134,14 @@ func runHelloWorldWorker(ctx context.Context, podTemplateSpec corev1.PodTemplate
134134
w.RegisterWorkflow(helloWorld)
135135
w.RegisterActivity(getSubject)
136136
w.RegisterActivity(sleep)
137-
err = w.Start()
137+
138+
// Start the worker in a separate goroutine so that the stopFunc can be passed back to the caller via callback
139+
go func() {
140+
err = w.Start()
141+
if err != nil {
142+
callback(nil, err)
143+
}
144+
}()
138145
}
139146

140147
func setActivityTimeout(ctx workflow.Context, d time.Duration) workflow.Context {

0 commit comments

Comments
 (0)