Enable E2E Tests for Kubeflow Pipeline on ppc64le Architecture

.github/actions/kfp-cluster/action.yml

@@ -10,16 +10,12 @@ runs:
   using: "composite"
   steps:
     - name: Create k8s Kind Cluster
-      uses: container-tools/kind-action@v2
-      with:
-        cluster_name: kfp
-        kubectl_version: ${{ inputs.k8s_version }}
-        version: v0.25.0
-        node_image: kindest/node:${{ inputs.k8s_version }}
-
-    - name: Build images
       shell: bash
-      run: ./.github/resources/scripts/build-images.sh
+      run: kind create cluster --image quay.io/powercloud/kind-node:${{ inputs.k8s_version }} --config /home/ubuntu/pipelines/config.yaml
+
+    # - name: Build images
+    #   shell: bash
+    #   run: ./.github/resources/scripts/build-images.sh
 
     - name: Deploy KFP
       shell: bash

.github/workflows/e2e-test.yml

@@ -18,20 +18,15 @@ on:
 
 jobs:
   initialization-tests-v1:
-    runs-on: ubuntu-latest
+    runs-on: self-hosted
     strategy:
       matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
+        k8s_version: [ "v1.30.2", "v1.31.0" ]
     name: Initialization tests v1 - K8s ${{ matrix.k8s_version }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
 
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
       - name: Create KFP cluster
         uses: ./.github/actions/kfp-cluster
         with:
@@ -51,27 +46,25 @@ jobs:
           name: kfp-initialization-tests-v1-artifacts-k8s-${{ matrix.k8s_version }}
           path: /tmp/tmp.*/*
 
+      - name: Delete KFP cluster
+        shell: bash
+        run: kind delete cluster
+
   initialization-tests-v2:
-    runs-on: ubuntu-latest
+    runs-on: self-hosted
     strategy:
       matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
+        k8s_version: [ "v1.30.2", "v1.31.0" ]
     name: Initialization tests v2 - K8s ${{ matrix.k8s_version }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
 
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
       - name: Create KFP cluster
         uses: ./.github/actions/kfp-cluster
         with:
           k8s_version: ${{ matrix.k8s_version }}
 
-
       - name: Forward API port
         run: ./.github/resources/scripts/forward-port.sh "kubeflow" "ml-pipeline" 8888 8888
 
@@ -86,21 +79,20 @@ jobs:
           name: kfp-initialization-tests-v2-artifacts-k8s-${{ matrix.k8s_version }}
           path: /tmp/tmp.*/*
 
+      - name: Delete KFP cluster
+        shell: bash
+        run: kind delete cluster
+
   api-integration-tests-v1:
-    runs-on: ubuntu-latest
+    runs-on: self-hosted
     strategy:
       matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
+        k8s_version: [ "v1.30.2", "v1.31.0" ]
     name: API integration tests v1 - K8s ${{ matrix.k8s_version }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
 
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
       - name: Create KFP cluster
         uses: ./.github/actions/kfp-cluster
         with:
@@ -120,21 +112,20 @@ jobs:
           name: kfp-api-integration-tests-v1-artifacts-k8s-${{ matrix.k8s_version }}
           path: /tmp/tmp.*/*
 
+      - name: Delete KFP cluster
+        shell: bash
+        run: kind delete cluster
+
   api-integration-tests-v2:
-    runs-on: ubuntu-latest
+    runs-on: self-hosted
     strategy:
       matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
+        k8s_version: [ "v1.30.2", "v1.31.0" ]
     name: API integration tests v2 - K8s ${{ matrix.k8s_version }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
 
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
       - name: Create KFP cluster
         uses: ./.github/actions/kfp-cluster
         with:
@@ -154,81 +145,6 @@ jobs:
           name: kfp-api-integration-tests-v2-artifacts-k8s-${{ matrix.k8s_version }}
           path: /tmp/tmp.*/*
 
-  frontend-integration-test:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
-    name: Frontend Integration Tests - K8s ${{ matrix.k8s_version }}
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
-
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
-      - name: Create KFP cluster
-        uses: ./.github/actions/kfp-cluster
-        with:
-          k8s_version: ${{ matrix.k8s_version }}
-
-      - name: Forward API port
-        run: ./.github/resources/scripts/forward-port.sh "kubeflow" "ml-pipeline" 8888 8888
-
-      - name: Forward Frontend port
-        run: ./.github/resources/scripts/forward-port.sh "kubeflow" "ml-pipeline-ui" 3000 3000
-
-      - name: Build frontend integration tests image
-        working-directory: ./test/frontend-integration-test
-        run: docker build . -t kfp-frontend-integration-test:local
-
-      - name: Frontend integration tests
-        run: docker run --net=host kfp-frontend-integration-test:local --remote-run true
-
-      - name: Collect test results
-        if: always()
-        uses: actions/upload-artifact@v4
-        with:
-          name: kfp-frontend-integration-test-artifacts-k8s-${{ matrix.k8s_version }}
-          path: /tmp/tmp.*/*
-
-  basic-sample-tests:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        k8s_version: [ "v1.29.2", "v1.30.2", "v1.31.0" ]
-    name: Basic Sample Tests - K8s ${{ matrix.k8s_version }}
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
-
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: 3.9
-
-      - name: Create KFP cluster
-        uses: ./.github/actions/kfp-cluster
-        with:
-          k8s_version: ${{ matrix.k8s_version }}
-
-      - name: Forward API port
-        run: ./.github/resources/scripts/forward-port.sh "kubeflow" "ml-pipeline" 8888 8888
-
-      - name: Install prerequisites
-        run: pip3 install -r ./test/sample-test/requirements.txt
-
-      - name: Basic sample tests - sequential
-        run:  python3 ./test/sample-test/sample_test_launcher.py sample_test run_test --namespace kubeflow --test-name sequential --results-gcs-dir output
-
-      - name: Basic sample tests - exit_handler
-        run: python3 ./test/sample-test/sample_test_launcher.py sample_test run_test --namespace kubeflow --test-name exit_handler --results-gcs-dir output
-
-      - name: Collect test results
-        if: always()
-        uses: actions/upload-artifact@v4
-        with:
-          name: kfp-basic-sample-tests-artifacts-k8s-${{ matrix.k8s_version }}
-          path: /tmp/tmp.*/*
+      - name: Delete KFP cluster
+        shell: bash
+        run: kind delete cluster

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

@@ -39,7 +39,7 @@ type Saver interface {
 }
 
 type EventHandler interface {
-	AddEventHandler(handler cache.ResourceEventHandler)
+	AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error)
 }
 
 // PersistenceWorker is a generic worker to persist objects from a queue.

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

@@ -27,6 +27,16 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
+type FakeResourceEventHandlerRegistration struct{}
+
+func (h *FakeResourceEventHandlerRegistration) HasSynced() bool {
+	return true
+}
+
+func NewFakeResourceEventHandlerRegistration() *FakeResourceEventHandlerRegistration {
+	return &FakeResourceEventHandlerRegistration{}
+}
+
 type FakeEventHandler struct {
 	handler cache.ResourceEventHandler
 }
@@ -35,8 +45,9 @@ func NewFakeEventHandler() *FakeEventHandler {
 	return &FakeEventHandler{}
 }
 
-func (h *FakeEventHandler) AddEventHandler(handler cache.ResourceEventHandler) {
+func (h *FakeEventHandler) AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error) {
 	h.handler = handler
+	return NewFakeResourceEventHandlerRegistration(), nil
 }
 
 func TestPersistenceWorker_Success(t *testing.T) {
@@ -65,7 +76,7 @@ func TestPersistenceWorker_Success(t *testing.T) {
 		saver)
 
 	// Test
-	eventHandler.handler.OnAdd(workflow)
+	eventHandler.handler.OnAdd(workflow, true)
 	worker.processNextWorkItem()
 	assert.Equal(t, workflow, pipelineClient.GetWorkflow("MY_NAMESPACE", "MY_NAME"))
 	assert.Equal(t, 0, worker.Len())
@@ -95,7 +106,7 @@ func TestPersistenceWorker_NotFoundError(t *testing.T) {
 		saver)
 
 	// Test
-	eventHandler.handler.OnAdd(workflow)
+	eventHandler.handler.OnAdd(workflow, true)
 	worker.processNextWorkItem()
 	assert.Nil(t, pipelineClient.GetWorkflow("MY_NAMESPACE", "MY_NAME"))
 	assert.Equal(t, 0, worker.Len())
@@ -126,7 +137,7 @@ func TestPersistenceWorker_GetWorklowError(t *testing.T) {
 		saver)
 
 	// Test
-	eventHandler.handler.OnAdd(workflow)
+	eventHandler.handler.OnAdd(workflow, true)
 	worker.processNextWorkItem()
 	assert.Nil(t, pipelineClient.GetWorkflow("MY_NAMESPACE", "MY_NAME"))
 	assert.Equal(t, 1, worker.Len())
@@ -160,7 +171,7 @@ func TestPersistenceWorker_ReportWorkflowRetryableError(t *testing.T) {
 		saver)
 
 	// Test
-	eventHandler.handler.OnAdd(workflow)
+	eventHandler.handler.OnAdd(workflow, true)
 	worker.processNextWorkItem()
 	assert.Nil(t, pipelineClient.GetWorkflow("MY_NAMESPACE", "MY_NAME"))
 	assert.Equal(t, 1, worker.Len())
@@ -193,7 +204,7 @@ func TestPersistenceWorker_ReportWorkflowNonRetryableError(t *testing.T) {
 		saver)
 
 	// Test
-	eventHandler.handler.OnAdd(workflow)
+	eventHandler.handler.OnAdd(workflow, true)
 	worker.processNextWorkItem()
 	assert.Nil(t, pipelineClient.GetWorkflow("MY_NAMESPACE", "MY_NAME"))
 	assert.Equal(t, 0, worker.Len())

backend/src/apiserver/model/cron_schedule.go

@@ -0,0 +1,34 @@
+// Copyright 2024 The Kubeflow Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package model
+
+type CronSchedule struct {
+	// Time at which scheduling starts.
+	// If no start time is specified, the StartTime is the creation time of the schedule.
+	CronScheduleStartTimeInSec *int64 `gorm:"column:CronScheduleStartTimeInSec;"`
+
+	// Time at which scheduling ends.
+	// If no end time is specified, the EndTime is the end of time.
+	CronScheduleEndTimeInSec *int64 `gorm:"column:CronScheduleEndTimeInSec;"`
+
+	// Cron string describing when a workflow should be created within the
+	// time interval defined by StartTime and EndTime.
+	Cron *string `gorm:"column:Schedule;"`
+}
+
+func (c CronSchedule) IsEmpty() bool {
+	return (c.CronScheduleStartTimeInSec == nil || *c.CronScheduleStartTimeInSec == 0) &&
+		(c.CronScheduleEndTimeInSec == nil || *c.CronScheduleEndTimeInSec == 0) &&
+		(c.Cron == nil || *c.Cron == "")
+}

backend/src/apiserver/model/job.go

@@ -189,34 +189,6 @@ type Trigger struct {
 	PeriodicSchedule
 }
 
-type CronSchedule struct {
-	// Time at which scheduling starts.
-	// If no start time is specified, the StartTime is the creation time of the schedule.
-	CronScheduleStartTimeInSec *int64 `gorm:"column:CronScheduleStartTimeInSec;"`
-
-	// Time at which scheduling ends.
-	// If no end time is specified, the EndTime is the end of time.
-	CronScheduleEndTimeInSec *int64 `gorm:"column:CronScheduleEndTimeInSec;"`
-
-	// Cron string describing when a workflow should be created within the
-	// time interval defined by StartTime and EndTime.
-	Cron *string `gorm:"column:Schedule;"`
-}
-
-type PeriodicSchedule struct {
-	// Time at which scheduling starts.
-	// If no start time is specified, the StartTime is the creation time of the schedule.
-	PeriodicScheduleStartTimeInSec *int64 `gorm:"column:PeriodicScheduleStartTimeInSec;"`
-
-	// Time at which scheduling ends.
-	// If no end time is specified, the EndTime is the end of time.
-	PeriodicScheduleEndTimeInSec *int64 `gorm:"column:PeriodicScheduleEndTimeInSec;"`
-
-	// Interval describing when a workflow should be created within the
-	// time interval defined by StartTime and EndTime.
-	IntervalSecond *int64 `gorm:"column:IntervalSecond;"`
-}
-
 func (j Job) GetValueOfPrimaryKey() string {
 	return fmt.Sprint(j.UUID)
 }