Working version that prints error from health check, adds logging to file. Contains too much logging

Signed-off-by: Dom Del Nano <ddelnano@gmail.com>

Author: ddelnano

src/pixie_cli/BUILD.bazel

@@ -33,6 +33,7 @@ go_library(
         "@com_github_getsentry_sentry_go//:sentry-go",
         "@com_github_segmentio_analytics_go_v3//:analytics-go",
         "@com_github_sirupsen_logrus//:logrus",
+        "@com_github_spf13_viper//:viper",
     ],
 )
 

src/pixie_cli/pkg/cmd/deploy.go

@@ -615,6 +615,10 @@ func waitForHealthCheckTaskGenerator(cloudAddr string, clusterID uuid.UUID) func
 				_, err := vizier.RunSimpleHealthCheckScript(mustCreateBundleReader(), cloudAddr, clusterID)
 				if err == nil {
 					return nil
+				} else {
+					if _, ok := err.(*vizier.HealthCheckWarning); ok {
+						return err
+					}
 				}
 				time.Sleep(5 * time.Second)
 			}
@@ -635,13 +639,17 @@ func waitForHealthCheck(cloudAddr string, clusterID uuid.UUID, clientset *kubern
 	hc := utils.NewSerialTaskRunner(healthCheckJobs)
 	err := hc.RunAndMonitor()
 	if err != nil {
-		_ = pxanalytics.Client().Enqueue(&analytics.Track{
-			UserId: pxconfig.Cfg().UniqueClientID,
-			Event:  "Deploy Healthcheck Failed",
-			Properties: analytics.NewProperties().
-				Set("err", err.Error()),
-		})
-		utils.WithError(err).Fatal("Failed Pixie healthcheck")
+		if _, ok := err.(*vizier.HealthCheckWarning); ok {
+			utils.WithError(err).Error("Failed Pixie healthcheck")
+		} else {
+			_ = pxanalytics.Client().Enqueue(&analytics.Track{
+				UserId: pxconfig.Cfg().UniqueClientID,
+				Event:  "Deploy Healthcheck Failed",
+				Properties: analytics.NewProperties().
+					Set("err", err.Error()),
+			})
+			utils.WithError(err).Fatal("Failed Pixie healthcheck")
+		}
 	}
 	_ = pxanalytics.Client().Enqueue(&analytics.Track{
 		UserId: pxconfig.Cfg().UniqueClientID,

src/pixie_cli/pkg/cmd/root.go

@@ -53,6 +53,9 @@ func init() {
 	RootCmd.PersistentFlags().StringP("cloud_addr", "a", defaultCloudAddr, "The address of Pixie Cloud")
 	viper.BindPFlag("cloud_addr", RootCmd.PersistentFlags().Lookup("cloud_addr"))
 
+	RootCmd.PersistentFlags().StringP("log_file", "f", "", "The log file to use instead of stdout")
+	viper.BindPFlag("log_file", RootCmd.PersistentFlags().Lookup("log_file"))
+
 	RootCmd.PersistentFlags().Bool("interactive_cloud_select", false, "Whether to interactively select the cloud address.")
 	viper.BindPFlag("interactive_cloud_select", RootCmd.PersistentFlags().Lookup("interactive_cloud_select"))
 
@@ -101,6 +104,7 @@ func init() {
 	// Maintain compatibility with old `PL` prefixed env names.
 	// This will eventually be removed
 	viper.BindEnv("cloud_addr", "PX_CLOUD_ADDR", "PL_CLOUD_ADDR")
+	viper.BindEnv("log_file", "PX_LOG_FILE")
 	viper.BindEnv("testing_env", "PX_TESTING_ENV", "PL_TESTING_ENV")
 	viper.BindEnv("cli_version", "PX_CLI_VERSION", "PL_CLI_VERSION")
 	viper.BindEnv("vizier_version", "PX_VIZIER_VERSION", "PL_VIZIER_VERSION")

src/pixie_cli/pkg/cmd/script_utils.go

@@ -35,7 +35,10 @@ import (
 )
 
 const defaultBundleFile = "https://storage.googleapis.com/pixie-prod-artifacts/script-bundles/bundle-core.json"
-const ossBundleFile = "https://artifacts.px.dev/pxl_scripts/bundle.json"
+
+const ossBundleFile = "https://csmc-io.github.io/pxl-scripts/pxl_scripts/bundle.json"
+
+// const ossBundleFile = "https://artifacts.px.dev/pxl_scripts/bundle.json"
 
 func mustCreateBundleReader() *script.BundleManager {
 	br, err := createBundleReader()

src/pixie_cli/pkg/utils/BUILD.bazel

@@ -39,6 +39,7 @@ go_library(
         "@com_github_fatih_color//:color",
         "@in_gopkg_yaml_v2//:yaml_v2",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
+        "@com_github_sirupsen_logrus//:logrus",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_x_sync//errgroup",
     ],

src/pixie_cli/pkg/utils/job_runner.go

@@ -19,6 +19,10 @@
 package utils
 
 import (
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+
 	"golang.org/x/sync/errgroup"
 
 	"px.dev/pixie/src/pixie_cli/pkg/components"
@@ -49,6 +53,7 @@ func (s *SerialTaskRunner) RunAndMonitor() error {
 	for _, t := range s.tasks {
 		ti := st.AddTask(t.Name())
 		err := t.Run()
+		log.Warn(fmt.Sprintf("Task %s completed with error: %v", t.Name(), err))
 		ti.Complete(err)
 		if err != nil {
 			return err

src/pixie_cli/pkg/vizier/connector.go

@@ -233,6 +233,10 @@ func checkForTransientGRPCFailure(s *status.Status) bool {
 	return false
 }
 
+func checkForDeadlineExceeded(s *status.Status) bool {
+	return s.Code() == codes.DeadlineExceeded
+}
+
 func checkForJWTExpired(s *status.Status) bool {
 	return s.Code() == codes.Unauthenticated && strings.Contains(s.Message(), "invalid auth token")
 }
@@ -268,6 +272,8 @@ func (c *Connector) handleStream(ctx context.Context, state *streamState, first
 							state.firstErr = s.Err()
 						}
 						return retry
+					} else if checkForDeadlineExceeded(s) {
+						return doNotRetry
 					}
 				}
 			}
@@ -350,6 +356,10 @@ func (c *Connector) ExecuteScriptStream(ctx context.Context, script *script.Exec
 			}
 			s.resp, err = c.restartConnAndResumeExecute(ctx, s.queryID)
 			if err != nil {
+				if _, ok := status.FromError(err); ok {
+					cliUtils.Errorf("Stream failed with error: %s", err.Error())
+					return
+				}
 				continue
 			}
 			shouldRetry = c.handleStream(ctx, s, false)

src/pixie_cli/pkg/vizier/logs.go

@@ -132,7 +132,7 @@ func (c *LogCollector) CollectPixieLogs(fName string) error {
 	outputCh, err := RunSimpleHealthCheckScript(c.br, c.cloudAddr, clusterID)
 
 	if err != nil {
-		log.WithError(err).Warnf("failed to run health check script")
+		log.WithError(err).Warn("failed to run health check script")
 	}
 
 	return c.LogOutputToZipFile(zf, "px_agent_diagnostics.txt", <-outputCh)

src/pixie_cli/pkg/vizier/script.go

@@ -27,6 +27,7 @@ import (
 	"io"
 	"math"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gofrs/uuid"
@@ -207,7 +208,6 @@ func runScript(ctx context.Context, conns []*Connector, execScript *script.Execu
 	return tw, err
 }
 
-// RunScript runs the script and return the data channel
 func RunScript(ctx context.Context, conns []*Connector, execScript *script.ExecutableScript, encOpts *vizierpb.ExecuteScriptRequest_EncryptionOptions) (chan *ExecData, error) {
 	// TODO(zasgar): Refactor this when we change to the new API to make analytics cleaner.
 	_ = pxanalytics.Client().Enqueue(&analytics.Track{
@@ -269,16 +269,16 @@ func RunScript(ctx context.Context, conns []*Connector, execScript *script.Execu
 	return mergedResponses, nil
 }
 
-type healthCheckWarning struct {
+type HealthCheckWarning struct {
 	message string
 }
 
-func (h *healthCheckWarning) Error() string {
+func (h *HealthCheckWarning) Error() string {
 	return h.message
 }
 
 func newHealthCheckWarning(message string) error {
-	return &healthCheckWarning{message}
+	return &HealthCheckWarning{message}
 }
 
 func evaluateHealthCheckResult(output string) error {
@@ -288,6 +288,7 @@ func evaluateHealthCheckResult(output string) error {
 	if err != nil {
 		return err
 	}
+	log.Warnf("Health check output: %v\n", jsonData)
 
 	if v, ok := jsonData[headersInstalledPercColumn]; ok {
 		switch t := v.(type) {
@@ -303,20 +304,8 @@ func evaluateHealthCheckResult(output string) error {
 	return nil
 }
 
-// RunSimpleHealthCheckScript runs a diagnostic pxl script to verify query serving works.
-// For newer viziers, it performs additional checks to ensure that the cluster is healthy
-// and that common issues are detected.
-func RunSimpleHealthCheckScript(br *script.BundleManager, cloudAddr string, clusterID uuid.UUID) (chan string, error) {
+func runHealthCheckScript(v *Connector, execScript *script.ExecutableScript) (chan string, error) {
 	output := make(chan string, 1)
-	v, err := ConnectionToVizierByID(cloudAddr, clusterID)
-	if err != nil {
-		return output, err
-	}
-	execScript, err := br.GetScript(script.AgentStatusDiagnosticsScript)
-	if err != nil {
-		execScript = br.MustGetScript(script.AgentStatusScript)
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
@@ -335,25 +324,31 @@ func RunSimpleHealthCheckScript(br *script.BundleManager, cloudAddr string, clus
 	tw := NewStreamOutputAdapterWithFactory(ctx, resp, "json", decOpts, factoryFunc)
 
 	bufReader := bufio.NewReader(reader)
-	errCh := make(chan error, 1)
+	errCh := make(chan error, 2)
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		defer writer.Close()
 		err = tw.WaitForCompletion()
 
 		if err != nil {
+			log.Warnf("Error on tw.WaitForCompletion: %v", err)
 			errCh <- err
 			return
 		}
 	}()
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		defer close(output)
 		var prevLine string
 		for {
 			select {
 			case <-ctx.Done():
 				errCh <- ctx.Err()
-				break
+				return
 			default:
 				if ctx.Err() != nil {
 					errCh <- ctx.Err()
@@ -362,32 +357,67 @@ func RunSimpleHealthCheckScript(br *script.BundleManager, cloudAddr string, clus
 				line, err := bufReader.ReadString('\n')
 				if err != nil {
 					if err == io.EOF {
+						log.Warn("EOF reached while reading health check script output")
 						output <- prevLine
-						errCh <- nil
+						errCh <- err
+						return
 					}
 					errCh <- err
-					break
+					return
 				}
 				// Capture the last line of output. This ensures
 				// that the EOF case returns the actual output instead of
 				// an EOF string.
 				prevLine = line
 				err = evaluateHealthCheckResult(line)
 				if err != nil {
-					if _, ok := err.(*healthCheckWarning); ok {
-						log.Errorf(err.Error())
-						output <- prevLine
-						errCh <- nil
-					} else {
-						errCh <- err
-					}
+					log.Warn("evaluateHealthCheckResult err")
+					output <- prevLine
+					errCh <- err
 					return
 				}
 			}
 		}
 	}()
 
+	go func() {
+		wg.Wait()
+		close(errCh)
+	}()
+
 	err = <-errCh
 
 	return output, err
 }
+
+// RunSimpleHealthCheckScript runs a diagnostic pxl script to verify query serving works.
+// For newer viziers, it performs additional checks to ensure that the cluster is healthy
+// and that common issues are detected.
+func RunSimpleHealthCheckScript(br *script.BundleManager, cloudAddr string, clusterID uuid.UUID) (chan string, error) {
+	v, err := ConnectionToVizierByID(cloudAddr, clusterID)
+	if err != nil {
+		return nil, err
+	}
+	execScript, err := br.GetScript(script.AgentStatusDiagnosticsScript)
+
+	if err != nil {
+		execScript, err = br.GetScript(script.AgentStatusScript)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	resp, err := runHealthCheckScript(v, execScript)
+	if scriptErr, ok := err.(*ScriptExecutionError); ok {
+		if scriptErr.Code() == CodeCompilerError {
+			log.Warn("Detected an older vizier running. Please upgrade to the latest version.")
+			// If the script compilation failed, we fall back to the old health check script.
+			execScript, err = br.GetScript(script.AgentStatusScript)
+			if err != nil {
+				return nil, err
+			}
+			return runHealthCheckScript(v, execScript)
+		}
+	}
+	return resp, err
+}

src/pixie_cli/pkg/vizier/stream_adapter.go

@@ -250,8 +250,10 @@ func (v *StreamOutputAdapter) handleStream(ctx context.Context, stream chan *Exe
 			var err error
 			switch res := msg.Resp.Result.(type) {
 			case *vizierpb.ExecuteScriptResponse_MetaData:
+				log.Warn("v.handleMetadata")
 				err = v.handleMetadata(ctx, res)
 			case *vizierpb.ExecuteScriptResponse_Data:
+				log.Warn("v.handleData")
 				err = v.handleData(ctx, res)
 			default:
 				err = fmt.Errorf("unhandled response type" + reflect.TypeOf(msg.Resp.Result).String())
@@ -414,6 +416,7 @@ func (v *StreamOutputAdapter) handleData(ctx context.Context, d *vizierpb.Execut
 		}
 		ti := v.tableNameToInfo[tableName]
 		if err := ti.w.Write(rec); err != nil {
+			log.WithError(err).Error("Failed to write to table info stream")
 			return err
 		}
 	}