[backend] DeletePipeline does not clean PipelienVersions data from Minio

[difince]

When a Pipeline has Pipeline Versions for each version there is a record in the Object Store (Minio). On delete Pipeline all the records (Pipeline and its Versions) are deleted from the database(MySQL), but not all are deleted from Minio. Minio records for Pipeline Versions are not deleted.

Environment

Linux Ubuntu - 20.04.3 LTS
kind v0.11.1 go1.16.4 linux/amd64
kustomize version: {KustomizeVersion:3.2.0 GitCommit:a3103f1e62ddb5b696daa3fd359bb6f2e8333b49 BuildDate:2019-09-18T16:26:36Z GoOs:linux GoArch:amd64}
Docker Engine 20.10.12

• How did you deploy Kubeflow Pipelines (KFP)?

Full Kubeflow deployment
while ! kustomize build example | kubectl apply -f -; do echo "Retrying to apply resources"; sleep 10; done
with manifest commit

• KFP version: Kubeflow Pipelines apps/pipeline/upstream 1.7.0
  But the problem exists with the latest Pipeline version as well!
• KFP SDK version:

Steps to reproduce

1. Create a Pipeline with Pipeline versions
2. Delete the Pipeline
3. Check Minio - Minio records associated with the Pipeline Versions are not deleted.

Expected result

Minio records associated with the Pipeline Versions to be deleted as well

Materials and Reference

In delete Pipeline function there is a comment that clearly says - that the current implementation of this func supports only Pipelines with a single Version.
Today, multiple Pipeline Versions are supported and it is time to be handled appropriately when Deleting a Pipeline.

	// Delete pipeline file and DB entry.
	// Not fail the request if this step failed. A background run will do the cleanup.
	// https://github.com/kubeflow/pipelines/issues/388
	// TODO(jingzhang36): For now (before exposing version API), we have only 1
	// file with both pipeline and version pointing to it;  so it is ok to do
	// the deletion as follows. After exposing version API, we can have multiple
	// versions and hence multiple files, and we shall improve performance by
	// either using async deletion in order for this method to be non-blocking
	// or or exploring other performance optimization tools provided by gcs.
	err = r.objectStore.DeleteFile(r.objectStore.GetPipelineKey(fmt.Sprint(pipelineId)))
	if err != nil {
		glog.Errorf("%v", errors.Wrapf(err, "Failed to delete pipeline file for pipeline %v", pipelineId))
		return nil
	}
	err = r.pipelineStore.DeletePipeline(pipelineId)

---

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.

[difince]

/assign @difince

[zijianjoy]

/cc @chensun

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[github-actions]

This issue has been automatically closed because it has not had recent activity. Please comment "/reopen" to reopen it.