update persistent agence to only store the argo spec (#1634)

* update persistent agence to only store the argo spec

Currently when persisting the runs spawned from a job, PersistentAgent stores more information than needed into the pipeline manifest, and also miss the TypeMetadata. This resulted in storing lots of runtime information that's not needed.

Example WorkflowSpec
```
   "metadata":{
      "name":"fffpw4fh-2-2911767673",
      "namespace":"kubeflow",
      "selfLink":"/apis/argoproj.io/v1alpha1/namespaces/kubeflow/workflows/fffpw4fh-2-2911767673",
      "uid":"de23bd5c-a8c1-11e9-a176-42010a800233",
      "resourceVersion":"3975687",
      "generation":1,
      "creationTimestamp":"2019-07-17T18:37:02Z",
      "labels":{
         "scheduledworkflows.kubeflow.org/isOwnedByScheduledWorkflow":"true",
         "scheduledworkflows.kubeflow.org/scheduledWorkflowName":"fffpw4fh",
         "scheduledworkflows.kubeflow.org/workflowEpoch":"1563388612",
         "scheduledworkflows.kubeflow.org/workflowIndex":"2",
         "workflows.argoproj.io/phase":"Running"
      },
      "ownerReferences":[
         {
            "apiVersion":"kubeflow.org/v1beta1",
            "kind":"ScheduledWorkflow",
            "name":"fffpw4fh",
            "uid":"91039a28-a8c1-11e9-a176-42010a800233",
            "controller":true,
            "blockOwnerDeletion":true
         }
      ]
   },
```

* Update workflow_test.go

* Update workflow.go

* Update resource_manager_test.go

* Update resource_manager.go

* Update workflow_test.go

backend/src/apiserver/resource/resource_manager.go

@@ -443,7 +443,7 @@ func (r *ResourceManager) ReportWorkflowResource(workflow *util.Workflow) error
 			FinishedAtInSec:  workflow.FinishedAt(),
 			Conditions:       workflow.Condition(),
 			PipelineSpec: model.PipelineSpec{
-				WorkflowSpecManifest: workflow.GetSpec().ToStringForStore(),
+				WorkflowSpecManifest: workflow.GetWorkflowSpec().ToStringForStore(),
 			},
 			ResourceReferences: []*model.ResourceReference{
 				{

backend/src/apiserver/resource/resource_manager_test.go

@@ -879,7 +879,7 @@ func TestReportWorkflowResource_ScheduledWorkflowIDNotEmpty_Success(t *testing.T
 			ScheduledAtInSec: 0,
 			FinishedAtInSec:  0,
 			PipelineSpec: model.PipelineSpec{
-				WorkflowSpecManifest: workflow.GetSpec().ToStringForStore(),
+				WorkflowSpecManifest: workflow.GetWorkflowSpec().ToStringForStore(),
 			},
 			ResourceReferences: []*model.ResourceReference{
 				{

backend/src/common/util/workflow.go

@@ -161,10 +161,12 @@ func (w *Workflow) HasScheduledWorkflowAsParent() bool {
 	return containsScheduledWorkflow(w.Workflow.OwnerReferences)
 }
 
-func (w *Workflow) GetSpec() *Workflow {
-	spec := w.DeepCopy()
-	spec.Status = workflowapi.WorkflowStatus{}
-	return NewWorkflow(spec)
+func (w *Workflow) GetWorkflowSpec() *Workflow {
+	workflow := w.DeepCopy()
+	workflow.Status = workflowapi.WorkflowStatus{}
+	workflow.TypeMeta = metav1.TypeMeta{Kind: w.Kind, APIVersion: w.APIVersion}
+	workflow.ObjectMeta = metav1.ObjectMeta{Name: w.Name, GenerateName: w.GenerateName}
+	return NewWorkflow(workflow)
 }
 
 // OverrideName sets the name of a Workflow.

backend/src/common/util/workflow_test.go

@@ -271,7 +271,7 @@ func TestSetLabels(t *testing.T) {
 	assert.Equal(t, expected, workflow.Get())
 }
 
-func TestGetSpec(t *testing.T) {
+func TestGetWorkflowSpec(t *testing.T) {
 	workflow := NewWorkflow(&workflowapi.Workflow{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   "WORKFLOW_NAME",
@@ -292,7 +292,6 @@ func TestGetSpec(t *testing.T) {
 	expected := &workflowapi.Workflow{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   "WORKFLOW_NAME",
-			Labels: map[string]string{"key": "value"},
 		},
 		Spec: workflowapi.WorkflowSpec{
 			Arguments: workflowapi.Arguments{
@@ -303,7 +302,7 @@ func TestGetSpec(t *testing.T) {
 		},
 	}
 
-	assert.Equal(t, expected, workflow.GetSpec().Get())
+	assert.Equal(t, expected, workflow.GetWorkflowSpec().Get())
 }
 
 func TestVerifyParameters(t *testing.T) {