chore(deps): Update client-go and controller-runtime

This updates k8s.io/client-go to v0.30.1 and
sigs.k8s.io/controller-runtime to v0.18.6. Updating any higher requires
updating github.com/argoproj/argo-workflows/v3 which requires updating
to Go 1.23 which isn't widely available in base container images yet.

Some other dependencies were updated for dependency resolution to
succeed.

Signed-off-by: mprahl <mprahl@users.noreply.github.com>

Author: mprahl

.github/workflows/backend.yml

@@ -21,12 +21,12 @@ jobs:
   run-go-unittests:
     runs-on: ubuntu-latest
     steps:
-    - name: Install Go
-      uses: actions/setup-go@v4
-      with:
-        go-version: 1.21.x
     - name: Checkout code
       uses: actions/checkout@v4
+    - name: Install Go
+      uses: actions/setup-go@v5
+      with:
+        go-version-file: go.mod
     - name: "run go unit tests"
       run: go test -v -cover ./backend/...
   backend-integration:

.github/workflows/presubmit-backend.yml

@@ -20,9 +20,9 @@ jobs:
       uses: actions/checkout@v2
 
     - name: Set up Go
-      uses: actions/setup-go@v3
+      uses: actions/setup-go@v5
       with:
-        go-version: '1.21'
+        go-version-file: go.mod
 
     - name: Run Backend Tests
       run: ./test/presubmit-backend-test.sh

backend/Dockerfile

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 # 1. Build api server application
-FROM golang:1.21.7-bookworm as builder
+FROM golang:1.22.10-bookworm as builder
 RUN apt-get update && apt-get install -y cmake clang musl-dev openssl
 WORKDIR /go/src/github.com/kubeflow/pipelines
 

backend/Dockerfile.cacheserver

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 # Dockerfile for building the source code of cache_server
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 RUN apk update && apk upgrade && \
     apk add --no-cache bash git openssh gcc musl-dev

backend/Dockerfile.conformance

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 # Dockerfile for building the source code of conformance tests
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 RUN apk update && apk upgrade && \
     apk add --no-cache bash git openssh gcc musl-dev

backend/Dockerfile.driver

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 WORKDIR /go/src/github.com/kubeflow/pipelines
 

backend/Dockerfile.launcher

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 WORKDIR /go/src/github.com/kubeflow/pipelines
 

backend/Dockerfile.persistenceagent

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 WORKDIR /go/src/github.com/kubeflow/pipelines
 

backend/Dockerfile.scheduledworkflow

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 WORKDIR /go/src/github.com/kubeflow/pipelines
 

backend/Dockerfile.viewercontroller

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.21.7-alpine3.19 as builder
+FROM golang:1.22.10-alpine3.21 as builder
 
 RUN apk update && apk upgrade
 RUN apk add --no-cache git gcc musl-dev

backend/api/Dockerfile

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 # Generate client code (go & json) from API protocol buffers
-FROM golang:1.21 as generator
+FROM golang:1.22 as generator
 ENV GRPC_GATEWAY_VERSION v1.9.6
 ENV GO_SWAGGER_VERSION v0.18.0
 ENV GOLANG_PROTOBUF_VERSION v1.5.1

backend/src/agent/persistence/client/scheduled_workflow_client.go

@@ -38,8 +38,8 @@ func NewScheduledWorkflowClient(informer v1beta1.ScheduledWorkflowInformer) *Sch
 }
 
 // AddEventHandler adds an event handler.
-func (c *ScheduledWorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) {
-	c.informer.Informer().AddEventHandler(funcs)
+func (c *ScheduledWorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) (cache.ResourceEventHandlerRegistration, error) {
+	return c.informer.Informer().AddEventHandler(funcs)
 }
 
 // HasSynced returns true if the shared informer's store has synced.

backend/src/agent/persistence/client/workflow_client.go

@@ -37,8 +37,8 @@ func NewWorkflowClient(informer util.ExecutionInformer) *WorkflowClient {
 }
 
 // AddEventHandler adds an event handler.
-func (c *WorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) {
-	c.informer.AddEventHandler(funcs)
+func (c *WorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) (cache.ResourceEventHandlerRegistration, error) {
+	return c.informer.AddEventHandler(funcs)
 }
 
 // HasSynced returns true if the shared informer's store has synced.

backend/src/agent/persistence/main.go

@@ -131,11 +131,14 @@ func main() {
 		log.Fatalf("Error creating ML pipeline API Server client: %v", err)
 	}
 
-	controller := NewPersistenceAgent(
+	controller, err := NewPersistenceAgent(
 		swfInformerFactory,
 		execInformer,
 		pipelineClient,
 		util.NewRealTime())
+	if err != nil {
+		log.Fatalf("Failed to instantiate the controller: %v", err)
+	}
 
 	go swfInformerFactory.Start(stopCh)
 	go execInformer.InformerFactoryStart(stopCh)

backend/src/agent/persistence/persistence_agent.go

@@ -46,7 +46,8 @@ func NewPersistenceAgent(
 	swfInformerFactory swfinformers.SharedInformerFactory,
 	execInformer util.ExecutionInformer,
 	pipelineClient *client.PipelineClient,
-	time util.TimeInterface) *PersistenceAgent {
+	time util.TimeInterface,
+) (*PersistenceAgent, error) {
 	// obtain references to shared informers
 	swfInformer := swfInformerFactory.Scheduledworkflow().V1beta1().ScheduledWorkflows()
 
@@ -57,12 +58,18 @@ func NewPersistenceAgent(
 	swfClient := client.NewScheduledWorkflowClient(swfInformer)
 	workflowClient := client.NewWorkflowClient(execInformer)
 
-	swfWorker := worker.NewPersistenceWorker(time, swfregister.Kind, swfInformer.Informer(), true,
+	swfWorker, err := worker.NewPersistenceWorker(time, swfregister.Kind, swfInformer.Informer(), true,
 		worker.NewScheduledWorkflowSaver(swfClient, pipelineClient))
+	if err != nil {
+		return nil, err
+	}
 
-	workflowWorker := worker.NewPersistenceWorker(time, workflowregister.WorkflowKind,
+	workflowWorker, err := worker.NewPersistenceWorker(time, workflowregister.WorkflowKind,
 		execInformer, true,
 		worker.NewWorkflowSaver(workflowClient, pipelineClient, ttlSecondsAfterWorkflowFinish))
+	if err != nil {
+		return nil, err
+	}
 
 	agent := &PersistenceAgent{
 		swfClient:      swfClient,
@@ -73,7 +80,7 @@ func NewPersistenceAgent(
 
 	log.Info("Setting up event handlers")
 
-	return agent
+	return agent, nil
 }
 
 // Run will set up the event handlers for types we are interested in, as well

backend/src/agent/persistence/worker/persistence_worker.go

@@ -62,7 +62,8 @@ func NewPersistenceWorker(
 	name string,
 	eventHandler util.ExecutionInformerEventHandler,
 	enforceRequeueDelays bool,
-	saver Saver) *PersistenceWorker {
+	saver Saver,
+) (*PersistenceWorker, error) {
 	worker := &PersistenceWorker{
 		workqueue: workqueue.NewNamedRateLimitingQueue(
 			workqueue.NewItemExponentialFailureRateLimiter(DefaultJobBackOff, MaxJobBackOff), name),
@@ -74,15 +75,18 @@ func NewPersistenceWorker(
 	log.Info("Setting up event handlers")
 
 	// Set up an event handler for when the Scheduled Workflow changes
-	eventHandler.AddEventHandler(&cache.ResourceEventHandlerFuncs{
+	_, err := eventHandler.AddEventHandler(&cache.ResourceEventHandlerFuncs{
 		AddFunc: worker.enqueue,
 		UpdateFunc: func(old, new interface{}) {
 			worker.enqueue(new)
 		},
 		DeleteFunc: worker.enqueueForDelete,
 	})
+	if err != nil {
+		return nil, err
+	}
 
-	return worker
+	return worker, nil
 }
 
 func (p *PersistenceWorker) Shutdown() {

backend/src/agent/persistence/worker/persistence_worker_test.go

@@ -22,6 +22,7 @@ import (
 	client "github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"k8s.io/client-go/tools/cache"
@@ -68,12 +69,13 @@ func TestPersistenceWorker_Success(t *testing.T) {
 	// Set up peristence worker
 	saver := NewWorkflowSaver(workflowClient, pipelineClient, 100)
 	eventHandler := NewFakeEventHandler()
-	worker := NewPersistenceWorker(
+	worker, err := NewPersistenceWorker(
 		util.NewFakeTimeForEpoch(),
 		"PERSISTENCE_WORKER",
 		eventHandler,
 		false,
 		saver)
+	require.NoError(t, err)
 
 	// Test
 	eventHandler.handler.OnAdd(workflow, true)
@@ -98,12 +100,13 @@ func TestPersistenceWorker_NotFoundError(t *testing.T) {
 	// Set up peristence worker
 	saver := NewWorkflowSaver(workflowClient, pipelineClient, 100)
 	eventHandler := NewFakeEventHandler()
-	worker := NewPersistenceWorker(
+	worker, err := NewPersistenceWorker(
 		util.NewFakeTimeForEpoch(),
 		"PERSISTENCE_WORKER",
 		eventHandler,
 		false,
 		saver)
+	require.NoError(t, err)
 
 	// Test
 	eventHandler.handler.OnAdd(workflow, true)
@@ -129,12 +132,13 @@ func TestPersistenceWorker_GetWorklowError(t *testing.T) {
 	// Set up peristence worker
 	saver := NewWorkflowSaver(workflowClient, pipelineClient, 100)
 	eventHandler := NewFakeEventHandler()
-	worker := NewPersistenceWorker(
+	worker, err := NewPersistenceWorker(
 		util.NewFakeTimeForEpoch(),
 		"PERSISTENCE_WORKER",
 		eventHandler,
 		false,
 		saver)
+	require.NoError(t, err)
 
 	// Test
 	eventHandler.handler.OnAdd(workflow, true)
@@ -163,12 +167,13 @@ func TestPersistenceWorker_ReportWorkflowRetryableError(t *testing.T) {
 	// Set up peristence worker
 	saver := NewWorkflowSaver(workflowClient, pipelineClient, 100)
 	eventHandler := NewFakeEventHandler()
-	worker := NewPersistenceWorker(
+	worker, err := NewPersistenceWorker(
 		util.NewFakeTimeForEpoch(),
 		"PERSISTENCE_WORKER",
 		eventHandler,
 		false,
 		saver)
+	require.NoError(t, err)
 
 	// Test
 	eventHandler.handler.OnAdd(workflow, true)
@@ -196,12 +201,13 @@ func TestPersistenceWorker_ReportWorkflowNonRetryableError(t *testing.T) {
 	// Set up peristence worker
 	saver := NewWorkflowSaver(workflowClient, pipelineClient, 100)
 	eventHandler := NewFakeEventHandler()
-	worker := NewPersistenceWorker(
+	worker, err := NewPersistenceWorker(
 		util.NewFakeTimeForEpoch(),
 		"PERSISTENCE_WORKER",
 		eventHandler,
 		false,
 		saver)
+	require.NoError(t, err)
 
 	// Test
 	eventHandler.handler.OnAdd(workflow, true)

backend/src/crd/controller/scheduledworkflow/client/swf_client.go

@@ -40,8 +40,8 @@ func NewScheduledWorkflowClient(clientSet swfclientset.Interface,
 }
 
 // AddEventHandler adds an event handler.
-func (p *ScheduledWorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) {
-	p.informer.Informer().AddEventHandler(funcs)
+func (p *ScheduledWorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) (cache.ResourceEventHandlerRegistration, error) {
+	return p.informer.Informer().AddEventHandler(funcs)
 }
 
 // HasSynced returns true if the shared informer's store has synced.

backend/src/crd/controller/scheduledworkflow/client/workflow_client.go

@@ -44,8 +44,8 @@ func NewWorkflowClient(clientSet commonutil.ExecutionClient,
 }
 
 // AddEventHandler adds an event handler.
-func (p *WorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) {
-	p.informer.AddEventHandler(funcs)
+func (p *WorkflowClient) AddEventHandler(funcs *cache.ResourceEventHandlerFuncs) (cache.ResourceEventHandlerRegistration, error) {
+	return p.informer.AddEventHandler(funcs)
 }
 
 // HasSynced returns true if the shared informer's store has synced.

backend/src/crd/controller/scheduledworkflow/controller.go

@@ -83,8 +83,8 @@ func NewController(
 	swfInformerFactory swfinformers.SharedInformerFactory,
 	executionInformer commonutil.ExecutionInformer,
 	time commonutil.TimeInterface,
-	location *time.Location) *Controller {
-
+	location *time.Location,
+) (*Controller, error) {
 	// obtain references to shared informers
 	swfInformer := swfInformerFactory.Scheduledworkflow().V1beta1().ScheduledWorkflows()
 
@@ -112,21 +112,24 @@ func NewController(
 	log.Info("Setting up event handlers")
 
 	// Set up an event handler for when the Scheduled Workflow changes
-	controller.swfClient.AddEventHandler(&cache.ResourceEventHandlerFuncs{
+	_, err := controller.swfClient.AddEventHandler(&cache.ResourceEventHandlerFuncs{
 		AddFunc: controller.enqueueScheduledWorkflow,
 		UpdateFunc: func(old, new interface{}) {
 			controller.enqueueScheduledWorkflow(new)
 		},
 		DeleteFunc: controller.enqueueScheduledWorkflowForDelete,
 	})
+	if err != nil {
+		return nil, err
+	}
 
 	// Set up an event handler for when WorkflowHistory resources change. This
 	// handler will lookup the owner of the given WorkflowHistory, and if it is
 	// owned by a ScheduledWorkflow, it will enqueue that ScheduledWorkflow for
 	// processing. This way, we don't need to implement custom logic for
 	// handling WorkflowHistory resources. More info on this pattern:
 	// https://github.com/kubernetes/community/blob/8cafef897a22026d42f5e5bb3f104febe7e29830/contributors/devel/controllers.md
-	controller.workflowClient.AddEventHandler(&cache.ResourceEventHandlerFuncs{
+	_, err = controller.workflowClient.AddEventHandler(&cache.ResourceEventHandlerFuncs{
 		AddFunc: controller.handleWorkflow,
 		UpdateFunc: func(old, new interface{}) {
 			newWorkflow := new.(*workflowapi.Workflow)
@@ -140,8 +143,11 @@ func NewController(
 		},
 		DeleteFunc: controller.handleWorkflow,
 	})
+	if err != nil {
+		return nil, err
+	}
 
-	return controller
+	return controller, nil
 }
 
 // Run will set up the event handlers for types we are interested in, as well

backend/src/crd/controller/scheduledworkflow/main.go

@@ -85,14 +85,17 @@ func main() {
 		scheduleInformerFactory = swfinformers.NewFilteredSharedInformerFactory(scheduleClient, time.Second*30, namespace, nil)
 	}
 
-	controller := NewController(
+	controller, err := NewController(
 		kubeClient,
 		scheduleClient,
 		execClient,
 		scheduleInformerFactory,
 		execInformer,
 		commonutil.NewRealTime(),
 		location)
+	if err != nil {
+		log.Fatalf("Failed to instantiate the controller: %v", err)
+	}
 
 	go scheduleInformerFactory.Start(stopCh)
 	go execInformer.InformerFactoryStart(stopCh)