feat: added metrics to workflow_executions database related code

db/sql/20201116105825_add_metrics_to_workflows.sql

@@ -0,0 +1,9 @@
+-- +goose Up
+-- SQL in this section is executed when the migration is applied.
+ALTER TABLE workflow_executions ADD COLUMN metrics JSONB;
+UPDATE workflow_executions SET metrics = '{}'::JSONB;
+ALTER TABLE workflow_executions ALTER COLUMN metrics SET NOT NULL;
+
+-- +goose Down
+-- SQL in this section is executed when the migration is rolled back.
+ALTER TABLE workflow_executions DROP COLUMN metrics;

pkg/workflow_execution.go

@@ -854,6 +854,7 @@ func (c *Client) createWorkflowExecutionDB(namespace string, workflowExecution *
 			"parameters":                   string(parametersJSON),
 			"is_archived":                  false,
 			"labels":                       workflowExecution.Labels,
+			"metrics":                      workflowExecution.Metrics,
 		}).
 		Suffix("RETURNING id").
 		RunWith(c.DB).
@@ -917,6 +918,7 @@ func (c *Client) CronStartWorkflowExecutionStatisticInsert(namespace, uid string
 			"cron_workflow_id":             cronWorkflow.ID,
 			"parameters":                   string(parametersJSON),
 			"labels":                       cronWorkflow.Labels,
+			"metrics":                      types.JSONLabels{},
 		}).
 		Suffix("RETURNING id").
 		RunWith(c.DB).

pkg/workflow_execution_types.go

@@ -25,6 +25,7 @@ type WorkflowExecution struct {
 	FinishedAt       *time.Time        `db:"finished_at"`
 	WorkflowTemplate *WorkflowTemplate `db:"workflow_template"`
 	Labels           types.JSONLabels
+	Metrics          types.JSONLabels
 	ArgoWorkflow     *wfv1.Workflow
 }
 
@@ -105,7 +106,18 @@ func (we *WorkflowExecution) GetParameterValue(name string) *string {
 // getWorkflowExecutionColumns returns all of the columns for workflowExecution modified by alias, destination.
 // see formatColumnSelect
 func getWorkflowExecutionColumns(aliasAndDestination ...string) []string {
-	columns := []string{"id", "created_at", "uid", "name", "parameters", "phase", "started_at", "finished_at", "labels"}
+	columns := []string{
+		"id",
+		"created_at",
+		"uid",
+		"name",
+		"parameters",
+		"phase",
+		"started_at",
+		"finished_at",
+		"labels",
+		"metrics",
+	}
 	return sql.FormatColumnSelect(columns, aliasAndDestination...)
 }
 
@@ -119,6 +131,7 @@ func getWorkflowExecutionColumnsMap(camelCase bool) map[string]string {
 		"parameters": "parameters",
 		"phase":      "phase",
 		"labels":     "labels",
+		"metrics":    "metrics",
 	}
 
 	if camelCase {