fix: restart watcher when concurrency backend changes

Author: chmouel

pkg/params/config_sync.go

@@ -13,6 +13,10 @@ import (
 	"knative.dev/pkg/system"
 )
 
+var terminateProcessForConfigChange = func() {
+	_ = syscall.Kill(os.Getpid(), syscall.SIGTERM)
+}
+
 func StartConfigSync(ctx context.Context, run *Run) {
 	// init pac config
 	_ = run.UpdatePacConfig(ctx)
@@ -28,7 +32,19 @@ func StartConfigSync(ctx context.Context, run *Run) {
 			// nothing to do
 		},
 		UpdateFunc: func(_, _ any) {
-			_ = run.UpdatePacConfig(ctx)
+			oldBackend, newBackend, changed, err := updatePacConfigAndDetectBackendChange(ctx, run)
+			if err != nil {
+				return
+			}
+			if changed {
+				if run.Clients.Log != nil {
+					run.Clients.Log.Infof(
+						"concurrency-backend changed from %q to %q; restarting process so the queue backend is recreated",
+						oldBackend, newBackend,
+					)
+				}
+				terminateProcessForConfigChange()
+			}
 		},
 		DeleteFunc: func(_ any) {
 			// nothing to do
@@ -46,3 +62,13 @@ func StartConfigSync(ctx context.Context, run *Run) {
 	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
 	<-sig
 }
+
+func updatePacConfigAndDetectBackendChange(ctx context.Context, run *Run) (string, string, bool, error) {
+	oldBackend := run.Info.GetPacOpts().ConcurrencyBackend
+	if err := run.UpdatePacConfig(ctx); err != nil {
+		return oldBackend, oldBackend, false, err
+	}
+
+	newBackend := run.Info.GetPacOpts().ConcurrencyBackend
+	return oldBackend, newBackend, oldBackend != "" && newBackend != "" && oldBackend != newBackend, nil
+}

pkg/params/config_sync_test.go

@@ -0,0 +1,89 @@
+package params
+
+import (
+	"testing"
+
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/consoleui"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/clients"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/info"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/settings"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubefake "k8s.io/client-go/kubernetes/fake"
+	rtesting "knative.dev/pkg/reconciler/testing"
+
+	"go.uber.org/zap"
+	"gotest.tools/v3/assert"
+)
+
+func TestUpdatePacConfigAndDetectBackendChange(t *testing.T) {
+	tests := []struct {
+		name           string
+		initialBackend string
+		configData     map[string]string
+		wantOld        string
+		wantNew        string
+		wantChanged    bool
+	}{
+		{
+			name:           "detects backend change",
+			initialBackend: settings.ConcurrencyBackendMemory,
+			configData: map[string]string{
+				"concurrency-backend":  settings.ConcurrencyBackendLease,
+				"tekton-dashboard-url": "https://dashboard.example.test",
+			},
+			wantOld:     settings.ConcurrencyBackendMemory,
+			wantNew:     settings.ConcurrencyBackendLease,
+			wantChanged: true,
+		},
+		{
+			name:           "ignores unchanged backend",
+			initialBackend: settings.ConcurrencyBackendLease,
+			configData: map[string]string{
+				"concurrency-backend":  settings.ConcurrencyBackendLease,
+				"tekton-dashboard-url": "https://dashboard.example.test",
+			},
+			wantOld:     settings.ConcurrencyBackendLease,
+			wantNew:     settings.ConcurrencyBackendLease,
+			wantChanged: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, _ := rtesting.SetupFakeContext(t)
+			ctx = info.StoreNS(ctx, "pac")
+
+			run := &Run{
+				Clients: clients.Clients{
+					Kube: kubefake.NewSimpleClientset(&corev1.ConfigMap{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "pac-config",
+							Namespace: "pac",
+						},
+						Data: tt.configData,
+					}),
+					Log: zap.NewNop().Sugar(),
+				},
+				Info: info.Info{
+					Pac: &info.PacOpts{
+						Settings: settings.Settings{
+							ConcurrencyBackend: tt.initialBackend,
+						},
+					},
+					Controller: &info.ControllerInfo{
+						Configmap: "pac-config",
+					},
+				},
+			}
+			run.Clients.InitClients()
+			run.Clients.SetConsoleUI(consoleui.FallBackConsole{})
+
+			oldBackend, newBackend, changed, err := updatePacConfigAndDetectBackendChange(ctx, run)
+			assert.NilError(t, err)
+			assert.Equal(t, oldBackend, tt.wantOld)
+			assert.Equal(t, newBackend, tt.wantNew)
+			assert.Equal(t, changed, tt.wantChanged)
+		})
+	}
+}