feat(backend): implement logs as artifacts + CI updates (#11809)

* feat(backend): implement logs as artifacts

Signed-off-by: droctothorpe <[email protected]>
Co-authored-by: cmdevoto <[email protected]>

* Address feedback and update golangci-lint

Signed-off-by: droctothorpe <[email protected]>

* Implement flag

Signed-off-by: droctothorpe <[email protected]>
Co-authored-by: carter.fendley <[email protected]>

* Handle logs in kubernetesPlatformOps

Signed-off-by: droctothorpe <[email protected]>

* Broaden sdk execution test filter

Signed-off-by: droctothorpe <[email protected]>

* Delete MinIO PVC at end of SDK execution tests

Signed-off-by: droctothorpe <[email protected]>

* Emit more logs from sdk execution tests

Signed-off-by: droctothorpe <[email protected]>

---------

Signed-off-by: droctothorpe <[email protected]>
Co-authored-by: cmdevoto <[email protected]>
Co-authored-by: carter.fendley <[email protected]>

Author: 3 people

.github/workflows/kfp-kubernetes-execution-tests.yml

@@ -11,6 +11,9 @@ on:
       - 'sdk/python/**'
       - 'api/v2alpha1/**'
       - 'kubernetes_platform/**'
+      - 'backend/**'
+      - '!**/*.md'
+      - '!**/OWNERS'
 
 jobs:
   kfp-kubernetes-execution-tests:

.github/workflows/sdk-execution.yml

@@ -10,7 +10,9 @@ on:
       - '.github/resources/**'
       - 'sdk/python/**'
       - 'api/v2alpha1/**'
-      - 'backend/src/v2/compiler/**'
+      - 'backend/**'
+      - '!**/*.md'
+      - '!**/OWNERS'
 
 jobs:
   sdk-execution-tests:
@@ -72,7 +74,7 @@ jobs:
         run: |
           export KFP_ENDPOINT="http://localhost:8888"
           export TIMEOUT_SECONDS=2700
-          pytest -v -n 5 ./test/sdk-execution-tests/sdk_execution_tests.py
+          pytest -v -n 5 -rP ./test/sdk-execution-tests/sdk_execution_tests.py
         continue-on-error: true
 
       - name: Collect failed logs

.pre-commit-config.yaml

@@ -55,7 +55,7 @@ repos:
 
   # Golang pre-submit hooks
   - repo: https://github.com/golangci/golangci-lint
-    rev: v1.52.2
+    rev: v1.64.8
     hooks:
       - id: golangci-lint
         name: golangci-lint

backend/src/v2/cmd/driver/main.go

@@ -19,9 +19,14 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+
 	"github.com/kubeflow/pipelines/backend/src/apiserver/config/proxy"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 
+	"os"
+	"path/filepath"
+	"strconv"
+
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
@@ -30,9 +35,6 @@ import (
 	"github.com/kubeflow/pipelines/backend/src/v2/driver"
 	"github.com/kubeflow/pipelines/backend/src/v2/metadata"
 	"github.com/kubeflow/pipelines/kubernetes_platform/go/kubernetesplatform"
-	"os"
-	"path/filepath"
-	"strconv"
 )
 
 const (
@@ -77,9 +79,10 @@ var (
 	logLevel           = flag.String("log_level", "1", "The verbosity level to log.")
 
 	// proxy
-	httpProxy  = flag.String(httpProxyArg, unsetProxyArgValue, "The proxy for HTTP connections.")
-	httpsProxy = flag.String(httpsProxyArg, unsetProxyArgValue, "The proxy for HTTPS connections.")
-	noProxy    = flag.String(noProxyArg, unsetProxyArgValue, "Addresses that should ignore the proxy.")
+	httpProxy   = flag.String(httpProxyArg, unsetProxyArgValue, "The proxy for HTTP connections.")
+	httpsProxy  = flag.String(httpsProxyArg, unsetProxyArgValue, "The proxy for HTTPS connections.")
+	noProxy     = flag.String(noProxyArg, unsetProxyArgValue, "Addresses that should ignore the proxy.")
+	publishLogs = flag.String("publish_logs", "true", "Whether to publish component logs to the object store")
 )
 
 // func RootDAG(pipelineName string, runID string, component *pipelinespec.ComponentSpec, task *pipelinespec.PipelineTaskSpec, mlmd *metadata.Client) (*Execution, error) {
@@ -189,6 +192,7 @@ func drive() (err error) {
 		DAGExecutionID:   *dagExecutionID,
 		IterationIndex:   *iterationIndex,
 		PipelineLogLevel: *logLevel,
+		PublishLogs:      *publishLogs,
 	}
 	var execution *driver.Execution
 	var driverErr error

backend/src/v2/cmd/launcher-v2/main.go

@@ -42,6 +42,7 @@ var (
 	mlmdServerAddress = flag.String("mlmd_server_address", "", "The MLMD gRPC server address.")
 	mlmdServerPort    = flag.String("mlmd_server_port", "8080", "The MLMD gRPC server port.")
 	logLevel          = flag.String("log_level", "1", "The verbosity level to log.")
+	publishLogs       = flag.String("publish_logs", "true", "Whether to publish component logs to the object store")
 )
 
 func main() {
@@ -79,6 +80,7 @@ func run() error {
 		MLMDServerPort:    *mlmdServerPort,
 		PipelineName:      *pipelineName,
 		RunID:             *runID,
+		PublishLogs:       *publishLogs,
 	}
 
 	switch *executorType {

backend/src/v2/compiler/argocompiler/container.go

@@ -16,12 +16,13 @@ package argocompiler
 
 import (
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/apiserver/config/proxy"
-	"k8s.io/apimachinery/pkg/util/intstr"
 	"os"
 	"strconv"
 	"strings"
 
+	"github.com/kubeflow/pipelines/backend/src/apiserver/config/proxy"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
 	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/jsonpb"
@@ -46,6 +47,7 @@ const (
 	DriverCommandEnvVar      = "V2_DRIVER_COMMAND"
 	PipelineRunAsUserEnvVar  = "PIPELINE_RUN_AS_USER"
 	PipelineLogLevelEnvVar   = "PIPELINE_LOG_LEVEL"
+	PublishLogsEnvVar        = "PUBLISH_LOGS"
 	gcsScratchLocation       = "/gcs"
 	gcsScratchName           = "gcs-scratch"
 	s3ScratchLocation        = "/s3"
@@ -192,6 +194,9 @@ func (c *workflowCompiler) addContainerDriverTemplate() string {
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)
 	}
+	if value, ok := os.LookupEnv(PublishLogsEnvVar); ok {
+		args = append(args, "--publish_logs", value)
+	}
 
 	t := &wfapi.Template{
 		Name: name,
@@ -316,6 +321,9 @@ func (c *workflowCompiler) addContainerExecutorTemplate(name string, refName str
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)
 	}
+	if value, ok := os.LookupEnv(PublishLogsEnvVar); ok {
+		args = append(args, "--publish_logs", value)
+	}
 	executor := &wfapi.Template{
 		Name:          nameContainerImpl,
 		RetryStrategy: c.getTaskRetryStrategy(name),

backend/src/v2/compiler/argocompiler/dag.go

@@ -559,6 +559,9 @@ func (c *workflowCompiler) addDAGDriverTemplate() string {
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)
 	}
+	if value, ok := os.LookupEnv(PublishLogsEnvVar); ok {
+		args = append(args, "--publish_logs", value)
+	}
 
 	t := &wfapi.Template{
 		Name: name,

backend/src/v2/compiler/argocompiler/importer.go

@@ -85,6 +85,9 @@ func (c *workflowCompiler) addImporterTemplate() string {
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)
 	}
+	if value, ok := os.LookupEnv(PublishLogsEnvVar); ok {
+		args = append(args, "--publish_logs", value)
+	}
 	importerTemplate := &wfapi.Template{
 		Name: name,
 		Inputs: wfapi.Inputs{