Merge branch 'main' into feat-o11y-overhaul

Author: grutt

.github/workflows/sdk-typescript.yml

@@ -264,7 +264,7 @@ jobs:
       - name: Run e2e tests
         env:
           HATCHET_CLIENT_TLS_STRATEGY: none
-          HATCHET_CLIENT_WORKER_HEALTHCHECK_ENABLED: "true"
+          HATCHET_CLIENT_WORKER_HEALTHCHECK_ENABLED: "false"
           NODE_TLS_REJECT_UNAUTHORIZED: "0"
         run: |
           echo "Testing current SDK against engine ${{ env.LATEST_TAG }}"

cmd/hatchet-engine/engine/run.go

@@ -102,8 +102,6 @@ func Run(ctx context.Context, cf *loader.ConfigLoader, version string) error {
 		return fmt.Errorf("could not run with config: %w", err)
 	}
 
-	time.Sleep(server.Runtime.ShutdownWait)
-
 	l.Debug().Msgf("interrupt received, shutting down")
 
 	err = cleanup.Run()

cmd/hatchet-loadtest/do.go

@@ -4,12 +4,74 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os"
+	"path/filepath"
 	"time"
+
+	"github.com/vicanso/go-charts/v2"
 )
 
+type LatencySnapshot struct {
+	t       time.Time
+	latency time.Duration
+}
+
+type LatencyResult struct {
+	snapshots []LatencySnapshot
+}
+
+func (lr *LatencyResult) GeneratePlot(plotPath string, plotName string) error {
+	bytes, err := lr.PlotBytes(plotName)
+	if err != nil {
+		return err
+	}
+
+	// save to file
+	f, err := os.Create(filepath.Join(plotPath, fmt.Sprintf("%s_plot.png", plotName)))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	_, err = f.Write(bytes)
+	return err
+}
+
+func (lr *LatencyResult) PlotBytes(plotName string) ([]byte, error) {
+	if len(lr.snapshots) == 0 {
+		return nil, fmt.Errorf("no snapshots available")
+	}
+
+	xvals := make([]string, 0, len(lr.snapshots))
+	yvals := make([]float64, 0, len(lr.snapshots))
+
+	start := lr.snapshots[0].t
+
+	for _, s := range lr.snapshots {
+		elapsed := s.t.Sub(start).Seconds()
+		xvals = append(xvals, fmt.Sprintf("%.2f", elapsed))
+
+		latencyMs := float64(s.latency.Microseconds()) / 1000.0
+		yvals = append(yvals, latencyMs)
+	}
+
+	p, err := charts.LineRender(
+		[][]float64{yvals},
+		charts.TitleTextOptionFunc(fmt.Sprintf("Task %s (ms)", plotName)),
+		charts.XAxisDataOptionFunc(xvals),
+		charts.LegendLabelsOptionFunc([]string{"Latency"}),
+		charts.HeightOptionFunc(500),
+		charts.WidthOptionFunc(1000),
+	)
+	if err != nil {
+		return nil, err
+	}
+	return p.Bytes()
+}
+
 type avgResult struct {
-	count int64
-	avg   time.Duration
+	count         int64
+	avg           time.Duration
+	latencyResult LatencyResult
 }
 
 func do(config LoadTestConfig) error {
@@ -25,22 +87,28 @@ func do(config LoadTestConfig) error {
 	defer cancel()
 
 	ch := make(chan int64, 2)
-	durations := make(chan time.Duration, config.Events)
+	durations := make(chan executionEvent, config.Events)
 
 	// Compute running average for executed durations using a rolling average.
 	durationsResult := make(chan avgResult)
 	go func() {
 		var count int64
 		var avg time.Duration
+		var snapshots []LatencySnapshot
+
 		for d := range durations {
 			count++
 			if count == 1 {
-				avg = d
+				avg = d.duration
 			} else {
-				avg += (d - avg) / time.Duration(count)
+				avg += (d.duration - avg) / time.Duration(count)
 			}
+			snapshots = append(snapshots, LatencySnapshot{
+				t:       d.startedAt,
+				latency: d.duration,
+			})
 		}
-		durationsResult <- avgResult{count: count, avg: avg}
+		durationsResult <- avgResult{count: count, avg: avg, latencyResult: LatencyResult{snapshots: snapshots}}
 	}()
 
 	// Start worker and ensure it has time to register
@@ -77,15 +145,20 @@ func do(config LoadTestConfig) error {
 	go func() {
 		var count int64
 		var avg time.Duration
+		var snapshots []LatencySnapshot
 		for d := range scheduled {
 			count++
 			if count == 1 {
 				avg = d
 			} else {
 				avg += (d - avg) / time.Duration(count)
 			}
+			snapshots = append(snapshots, LatencySnapshot{
+				t:       time.Now(),
+				latency: d,
+			})
 		}
-		scheduledResult <- avgResult{count: count, avg: avg}
+		scheduledResult <- avgResult{count: count, avg: avg, latencyResult: LatencyResult{snapshots: snapshots}}
 	}()
 
 	emitted := emit(ctx, config.Namespace, config.Events, config.Duration, scheduled, config.PayloadSize)
@@ -118,7 +191,37 @@ func do(config LoadTestConfig) error {
 
 	log.Printf("ℹ️ final average duration per executed event: %s", finalDurationResult.avg)
 	log.Printf("ℹ️ final average scheduling time per event: %s", finalScheduledResult.avg)
-
+	if ShouldSendSlack() {
+		log.Printf("ℹ️ sending scheduling/duration plots to Slack")
+		slackSender := NewSlackSender("hatchet-staging-loadtest-us-west-2")
+		durationBytes, err := finalDurationResult.latencyResult.PlotBytes("duration")
+		if err != nil {
+			log.Printf("❌ failed to generate duration plot: %v ", err)
+		}
+		schedulingBytes, err := finalScheduledResult.latencyResult.PlotBytes("scheduling")
+		if err != nil {
+			log.Printf("❌ failed to generate scheduling plot: %v ", err)
+		}
+		err = slackSender.Send(durationBytes, schedulingBytes, finalDurationResult.avg, finalScheduledResult.avg)
+		if err != nil {
+			log.Printf("❌ failed to send duration plots to slack: %v ", err)
+		}
+		log.Printf("ℹ️ scheduling/duration successfully plots to Slack")
+	} else {
+		log.Printf("ℹ️ not all environment vars for sending plots to Slack enabled...skipping")
+	}
+	if config.PlotDir != "" {
+		log.Printf("ℹ️ exporting scheduling/duration snapshot data")
+		err := finalScheduledResult.latencyResult.GeneratePlot(config.PlotDir, "scheduling")
+		if err != nil {
+			return err
+		}
+		err = finalDurationResult.latencyResult.GeneratePlot(config.PlotDir, "duration")
+		if err != nil {
+			return err
+		}
+		log.Printf("ℹ️ exported scheduling/duration snapshot data")
+	}
 	if expected != executed {
 		log.Printf("⚠️ warning: pushed and executed counts do not match: expected=%d got=%d", expected, executed)
 	}

cmd/hatchet-loadtest/main.go

@@ -35,6 +35,7 @@ type LoadTestConfig struct {
 	RlLimit                  int
 	RlDurationUnit           string
 	AverageDurationThreshold time.Duration
+	PlotDir                  string
 }
 
 func main() {
@@ -83,7 +84,7 @@ func main() {
 	loadtest.Flags().StringVar(&config.RlDurationUnit, "rlDurationUnit", "second", "rlDurationUnit specifies the duration unit for the rate limit (second, minute, hour)")
 	loadtest.Flags().StringVarP(&logLevel, "level", "l", "info", "logLevel specifies the log level (debug, info, warn, error)")
 	loadtest.Flags().DurationVar(&config.AverageDurationThreshold, "averageDurationThreshold", 100*time.Millisecond, "averageDurationThreshold specifies the threshold for the average duration per executed event to be considered a success")
-
+	loadtest.Flags().StringVar(&config.PlotDir, "plotDirectory", "", "plotDirectory specifies where to put the generated plots for latency and task duration")
 	cmd := &cobra.Command{Use: "app"}
 	cmd.AddCommand(loadtest)
 	if err := cmd.Execute(); err != nil {

cmd/hatchet-loadtest/run.go

@@ -22,7 +22,12 @@ type stepOneOutput struct {
 	Message string `json:"message"`
 }
 
-func run(ctx context.Context, config LoadTestConfig, executions chan<- time.Duration) (int64, int64) {
+type executionEvent struct {
+	startedAt time.Time
+	duration  time.Duration
+}
+
+func run(ctx context.Context, config LoadTestConfig, executions chan<- executionEvent) (int64, int64) {
 	hatchet, err := v1.NewHatchetClient(
 		v1.Config{
 			Namespace: config.Namespace,
@@ -44,7 +49,7 @@ func run(ctx context.Context, config LoadTestConfig, executions chan<- time.Dura
 		l.Info().Msgf("executing %d took %s", input.ID, took)
 
 		mx.Lock()
-		executions <- took
+		executions <- executionEvent{input.CreatedAt, took}
 		// detect duplicate in executed slice
 		var duplicate bool
 		// for i := 0; i < len(executed)-1; i++ {

cmd/hatchet-loadtest/slack.go

@@ -0,0 +1,129 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/google/uuid"
+	"github.com/slack-go/slack"
+)
+
+type SlackSender struct {
+	s3Client *s3.Client
+	s3Bucket string
+	Token    string
+	Channel  string
+	Thread   string
+}
+
+func NewS3Client(ctx context.Context) (*s3.Client, error) {
+	cfg, err := config.LoadDefaultConfig(ctx,
+		config.WithRegion("us-west-2"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	client := s3.NewFromConfig(cfg)
+	return client, nil
+}
+
+func ShouldSendSlack() bool {
+	return (os.Getenv("SLACK_BOT_TOKEN") != "" &&
+		os.Getenv("SLACK_THREAD_TS") != "" &&
+		os.Getenv("SLACK_CHANNEL_ID") != "" &&
+		os.Getenv("AWS_ACCESS_KEY_ID") != "" &&
+		os.Getenv("AWS_SECRET_ACCESS_KEY") != "")
+}
+
+func NewSlackSender(s3Bucket string) *SlackSender {
+	s3Client, _ := NewS3Client(context.Background())
+	return &SlackSender{
+		s3Client: s3Client,
+		s3Bucket: s3Bucket,
+		Token:    os.Getenv("SLACK_BOT_TOKEN"),
+		Thread:   os.Getenv("SLACK_THREAD_TS"),
+		Channel:  os.Getenv("SLACK_CHANNEL_ID"),
+	}
+}
+
+func (s *SlackSender) SendMessage(durationPlotUrl string, schedulingPlotUrl string, avgDuration time.Duration, avgScheduling time.Duration) error {
+	text := fmt.Sprintf(
+		":star:Load test results:star:\nAverage task duration: %s\nAverage task scheduling: %s",
+		avgDuration.String(),
+		avgScheduling.String(),
+	)
+
+	section := slack.NewSectionBlock(
+		slack.NewTextBlockObject("mrkdwn", text, false, false),
+		nil,
+		nil,
+	)
+
+	image1 := slack.NewImageBlock(
+		schedulingPlotUrl,
+		"Scheduling test graph",
+		"",
+		nil,
+	)
+
+	image2 := slack.NewImageBlock(
+		durationPlotUrl,
+		"Duration test graph",
+		"",
+		nil,
+	)
+
+	blocks := []slack.Block{section, image1, image2}
+	client := slack.New(s.Token)
+	_, _, err := client.PostMessage(
+		s.Channel,
+		slack.MsgOptionBlocks(blocks...),
+		slack.MsgOptionTS(s.Thread), // 👈 this attaches it to a thread
+	)
+	return err
+}
+
+func (s *SlackSender) UploadS3(imageBytes []byte) (*string, error) {
+	key := fmt.Sprintf("%s-%s-%s", "loadtest-plot", uuid.New(), time.Now().Format("20060102150405"))
+	_, err := s.s3Client.PutObject(context.Background(), &s3.PutObjectInput{
+		Bucket: aws.String(s.s3Bucket),
+		Key:    &key,
+		Body:   bytes.NewReader(imageBytes),
+	})
+	if err != nil {
+		return nil, err
+	}
+	presigner := s3.NewPresignClient(s.s3Client)
+	req, err := presigner.PresignGetObject(context.Background(),
+		&s3.GetObjectInput{
+			Bucket: &s.s3Bucket,
+			Key:    &key,
+		},
+		s3.WithPresignExpires(time.Hour*24*7),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	uploadedUrl := req.URL
+	return &uploadedUrl, nil
+}
+
+func (s *SlackSender) Send(durationBytes []byte, schedulingBytes []byte, avgDuration time.Duration, avgScheduling time.Duration) error {
+	uploadedDurationFileUrl, err := s.UploadS3(durationBytes)
+	if err != nil {
+		return err
+	}
+	uploadedSchedulingFileUrl, err := s.UploadS3(schedulingBytes)
+	if err != nil {
+		return err
+	}
+	return s.SendMessage(*uploadedDurationFileUrl, *uploadedSchedulingFileUrl, avgDuration, avgScheduling)
+}

frontend/app/src/pages/main/v1/logs/components/logs-chart.tsx

@@ -136,12 +136,6 @@ export function LogsChart({ metrics, since, until, onZoom }: LogsChartProps) {
                 />
               }
             />
-            <Bar
-              dataKey="DEBUG"
-              stackId="normal"
-              fill={CHART_CONFIG.DEBUG.color}
-              isAnimationActive={false}
-            />
             <Bar
               dataKey="INFO"
               stackId="normal"
@@ -153,12 +147,21 @@ export function LogsChart({ metrics, since, until, onZoom }: LogsChartProps) {
               stackId="normal"
               fill={CHART_CONFIG.WARN.color}
               isAnimationActive={false}
+              minPointSize={(value) => (value != null && value > 0 ? 2 : 0)}
+            />
+            <Bar
+              dataKey="DEBUG"
+              stackId="normal"
+              fill={CHART_CONFIG.DEBUG.color}
+              isAnimationActive={false}
+              minPointSize={(value) => (value != null && value > 0 ? 2 : 0)}
             />
             <Bar
               dataKey="ERROR"
               stackId="error"
               fill={CHART_CONFIG.ERROR.color}
               isAnimationActive={false}
+              minPointSize={(value) => (value != null && value > 0 ? 2 : 0)}
             />
 
             {refAreaLeft && refAreaRight && (