record-tester: Isolate stream health alerts from all others

internal/app/recordtester/continuous_record_tester.go

@@ -130,11 +130,17 @@ func (crt *continuousRecordTester) sendPagerdutyEvent(rt IRecordTester, err erro
 	if crt.pagerDutyIntegrationKey == "" {
 		return
 	}
+	isStreamHealth := errors.As(err, &testers.StreamHealthError{})
 	severity, lopriPrefix, dedupKey := "error", "", fmt.Sprintf("cont-record-tester:%s", crt.host)
-	if crt.pagerDutyLowUrgency {
+	if crt.pagerDutyLowUrgency || isStreamHealth {
 		severity, lopriPrefix = "warning", "[LOPRI] "
 		dedupKey = "lopri-" + dedupKey
 	}
+	componentName := ":movie_camera: LIVE"
+	if isStreamHealth {
+		dedupKey = "stream-health-" + dedupKey
+		componentName = ":hospital: STREAM_HEALTH"
+	}
 	event := pagerduty.V2Event{
 		RoutingKey: crt.pagerDutyIntegrationKey,
 		Action:     "trigger",
@@ -152,7 +158,7 @@ func (crt *continuousRecordTester) sendPagerdutyEvent(rt IRecordTester, err erro
 		Source:    crt.host,
 		Component: crt.pagerDutyComponent,
 		Severity:  severity,
-		Summary:   fmt.Sprintf("%s:movie_camera: RECORD %s for `%s` error: %v", lopriPrefix, crt.pagerDutyComponent, crt.host, err),
+		Summary:   fmt.Sprintf("%s%s %s for `%s` error: %v", componentName, lopriPrefix, crt.pagerDutyComponent, crt.host, err),
 		Timestamp: time.Now().UTC().Format(time.RFC3339),
 	}
 	sid := rt.StreamID()

internal/app/recordtester/recordtester_app.go

@@ -185,7 +185,7 @@ func (rt *recordTester) Start(fileName string, testDuration, pauseDuration time.
 		}
 		if srerr != nil {
 			glog.Warningf("Streaming returned error err=%v", srerr)
-			return 3, err
+			return 3, srerr
 		}
 		stats, err := sr2.Stats()
 		if err != nil {
@@ -210,7 +210,7 @@ func (rt *recordTester) Start(fileName string, testDuration, pauseDuration time.
 			}
 			if srerr != nil {
 				glog.Warningf("Streaming second returned error err=%v", srerr)
-				return 3, err
+				return 3, srerr
 			}
 			stats, err := sr2.Stats()
 			if err != nil {

internal/app/vodtester/continuous_vod_tester.go

@@ -124,7 +124,7 @@ func (cvt *continuousVodTester) sendPagerdutyEvent(vt IVodTester, err error) {
 		Source:    cvt.host,
 		Component: cvt.pagerDutyComponent,
 		Severity:  severity,
-		Summary:   fmt.Sprintf("%s:movie_camera: VOD %s for `%s` error: %v", lopriPrefix, cvt.pagerDutyComponent, cvt.host, err),
+		Summary:   fmt.Sprintf("%s:vhs: VOD %s for `%s` error: %v", lopriPrefix, cvt.pagerDutyComponent, cvt.host, err),
 		Timestamp: time.Now().UTC().Format(time.RFC3339),
 	}
 	resp, err := pagerduty.ManageEvent(event)

internal/testers/stream_health.go

@@ -3,7 +3,6 @@ package testers
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"sort"
 	"strings"
@@ -22,6 +21,10 @@ type (
 	// analyzer client configured to connect there.
 	AnalyzerByRegion map[string]client.Analyzer
 
+	StreamHealthError struct {
+		Message string
+	}
+
 	streamHealth struct {
 		finite
 		streamID string
@@ -88,7 +91,7 @@ func (h *streamHealth) workerLoop(waitForTarget time.Duration) {
 
 			msg := fmt.Sprintf("Global Stream Health API: stream did not become healthy on global analyzers after `%s`: %s",
 				waitForTarget, strings.Join(aggErrs, "; "))
-			h.fatalEnd(errors.New(msg))
+			h.fatalEnd(StreamHealthError{msg})
 			return
 		}
 	}
@@ -147,3 +150,7 @@ func conditionsStatus(health *data.HealthStatus) []string {
 	}
 	return failed
 }
+
+func (e StreamHealthError) Error() string {
+	return e.Message
+}

internal/testers/streamer2.go

@@ -100,7 +100,7 @@ func (sr *streamer2) StartStreaming(sourceFileName string, rtmpIngestURL, mediaU
 		var (
 			ctx, cancel = context.WithCancel(sr.ctx)
 			testsDone   = onAnyDone(ctx, tests)
-			errs        = []string{}
+			errs        []error
 		)
 		defer cancel()
 		for {
@@ -111,16 +111,11 @@ func (sr *streamer2) StartStreaming(sourceFileName string, rtmpIngestURL, mediaU
 						sr.globalError = err
 						time.AfterFunc(waitForTarget, cancel)
 					}
-					errs = append(errs, err.Error())
+					errs = append(errs, err)
 				}
 			case <-ctx.Done():
-				if len(errs) > 0 {
-					msg := errs[0]
-					if len(errs) > 1 {
-						sortErrs(errs)
-						msg = "Multiple errors: " + strings.Join(errs, "; ")
-					}
-					sr.fatalEnd(errors.New(msg))
+				if err := joinErrors(errs); err != nil {
+					sr.fatalEnd(err)
 				}
 				return
 			}
@@ -152,10 +147,23 @@ func onAnyDone(ctx context.Context, finites []Finite) <-chan Finite {
 	return finished
 }
 
-func sortErrs(errs []string) {
+func joinErrors(errs []error) error {
+	if len(errs) == 0 {
+		return nil
+	} else if len(errs) == 1 {
+		return errs[0]
+	}
+	sortErrs(errs)
+	msgs := make([]string, len(errs))
+	for i, err := range errs {
+		msgs[i] = err.Error()
+	}
+	return fmt.Errorf("Multiple errors: %s", strings.Join(msgs, "; "))
+}
+
+func sortErrs(errs []error) {
 	sortIdx := func(idx int) int {
-		err := strings.ToLower(errs[idx])
-		if strings.Contains(err, "health") {
+		if errors.As(errs[idx], &StreamHealthError{}) {
 			// stream health errs should go last. they're never the root cause when
 			// there are multiple errors.
 			return 1
@@ -166,7 +174,7 @@ func sortErrs(errs []string) {
 		if si1, si2 := sortIdx(idx1), sortIdx(idx2); si1 != si2 {
 			return si1 < si2
 		}
-		return errs[idx1] < errs[idx2]
+		return errs[idx1].Error() < errs[idx2].Error()
 	})
 }