-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
581 additions
and
217 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "net/url" | ||
| "time" | ||
|
|
||
| "github.com/PagerDuty/go-pagerduty" | ||
|
|
||
| "github.com/golang/glog" | ||
| "github.com/livepeer/stream-tester/messenger" | ||
| ) | ||
|
|
||
| type ( | ||
| IContinuousTester interface { | ||
| // Start test. Blocks until error. | ||
| Start(start func(ctx context.Context) error, testDuration, pauseBetweenTests time.Duration) error | ||
| Cancel() | ||
| Done() <-chan struct{} | ||
| } | ||
|
|
||
| ContinuousTesterOptions struct { | ||
| PagerDutyIntegrationKey string | ||
| PagerDutyComponent string | ||
| PagerDutyLowUrgency bool | ||
| TesterOptions | ||
| } | ||
|
|
||
| continuousTester struct { | ||
| ctx context.Context | ||
| cancel context.CancelFunc | ||
| host string // API host being tested | ||
| pagerDutyIntegrationKey string | ||
| pagerDutyComponent string | ||
| pagerDutyLowUrgency bool | ||
| name string | ||
| } | ||
| ) | ||
|
|
||
| func NewContinuousTester(gctx context.Context, opts ContinuousTesterOptions, testerName string) IContinuousTester { | ||
| ctx, cancel := context.WithCancel(gctx) | ||
| server := opts.API.GetServer() | ||
| u, _ := url.Parse(server) | ||
| ct := &continuousTester{ | ||
| ctx: ctx, | ||
| cancel: cancel, | ||
| host: u.Host, | ||
| pagerDutyIntegrationKey: opts.PagerDutyIntegrationKey, | ||
| pagerDutyComponent: opts.PagerDutyComponent, | ||
| pagerDutyLowUrgency: opts.PagerDutyLowUrgency, | ||
| name: testerName, | ||
| } | ||
| return ct | ||
| } | ||
|
|
||
| func (ct *continuousTester) Start(start func(ctx context.Context) error, testDuration, pauseBetweenTests time.Duration) error { | ||
| messenger.SendMessage(fmt.Sprintf("Starting continuous %s test of %s", ct.name, ct.host)) | ||
| for { | ||
| msg := fmt.Sprintf(":arrow_right: Starting %s %s test to %s", testDuration, ct.name, ct.host) | ||
| messenger.SendMessage(msg) | ||
|
|
||
| ctx, cancel := context.WithTimeout(ct.ctx, testDuration) | ||
| err := start(ctx) | ||
| ctxErr := ctx.Err() | ||
| cancel() | ||
|
|
||
| if ct.ctx.Err() != nil { | ||
| messenger.SendMessage(fmt.Sprintf("Continuous test of %s on %s cancelled", ct.name, ct.host)) | ||
| return ct.ctx.Err() | ||
| } else if ctxErr != nil { | ||
| msg := fmt.Sprintf("Test of %s on %s timed out, potential deadlock! ctxErr=%q err=%q", ct.name, ct.host, ctxErr, err) | ||
| messenger.SendFatalMessage(msg) | ||
| } else if err != nil { | ||
| msg := fmt.Sprintf(":rotating_light: Test of %s on %s ended with err=%v", ct.name, ct.host, err) | ||
| messenger.SendFatalMessage(msg) | ||
| glog.Warning(msg) | ||
| ct.sendPagerdutyEvent(err) | ||
| } else { | ||
| msg := fmt.Sprintf(":white_check_mark: Test of %s on %s succeeded", ct.name, ct.host) | ||
| messenger.SendMessage(msg) | ||
| glog.Info(msg) | ||
| ct.sendPagerdutyEvent(nil) | ||
| } | ||
| glog.Infof("Waiting %s before next test of %s", pauseBetweenTests, ct.name) | ||
| select { | ||
| case <-ct.ctx.Done(): | ||
| messenger.SendMessage(fmt.Sprintf("Continuous test of %s on %s cancelled", ct.name, ct.host)) | ||
| return err | ||
| case <-time.After(pauseBetweenTests): | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (ct *continuousTester) sendPagerdutyEvent(err error) { | ||
| if ct.pagerDutyIntegrationKey == "" { | ||
| return | ||
| } | ||
| severity, lopriPrefix, dedupKey := "error", "", fmt.Sprintf("cont-%s-tester:%s", ct.name, ct.host) | ||
| if ct.pagerDutyLowUrgency { | ||
| severity, lopriPrefix = "warning", "[LOPRI] " | ||
| dedupKey = "lopri-" + dedupKey | ||
| } | ||
| event := pagerduty.V2Event{ | ||
| RoutingKey: ct.pagerDutyIntegrationKey, | ||
| Action: "trigger", | ||
| DedupKey: dedupKey, | ||
| } | ||
| if err == nil { | ||
| event.Action = "resolve" | ||
| _, err := pagerduty.ManageEvent(event) | ||
| if err != nil { | ||
| messenger.SendMessage(fmt.Sprintf("Error resolving PagerDuty event: %v", err)) | ||
| } | ||
| return | ||
| } | ||
| summary := fmt.Sprintf("%s:vhs: %s %s for `%s` error: %v", lopriPrefix, ct.name, ct.pagerDutyComponent, ct.host, err) | ||
| if len(summary) > 1024 { | ||
| summary = summary[:1021] + "..." | ||
| } | ||
| event.Payload = &pagerduty.V2Payload{ | ||
| Source: ct.host, | ||
| Component: ct.pagerDutyComponent, | ||
| Severity: severity, | ||
| Summary: summary, | ||
| Timestamp: time.Now().UTC().Format(time.RFC3339), | ||
| } | ||
| resp, err := pagerduty.ManageEvent(event) | ||
| if err != nil { | ||
| glog.Error(fmt.Errorf("PAGERDUTY Error: %w", err)) | ||
| messenger.SendFatalMessage(fmt.Sprintf("Error creating PagerDuty event: %v", err)) | ||
| } else { | ||
| glog.Infof("Incident status: %s message: %s", resp.Status, resp.Message) | ||
| } | ||
| } | ||
|
|
||
| func (ct *continuousTester) Cancel() { | ||
| ct.cancel() | ||
| } | ||
|
|
||
| func (ct *continuousTester) Done() <-chan struct{} { | ||
| return ct.ctx.Done() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/golang/glog" | ||
| "github.com/livepeer/go-api-client" | ||
| "time" | ||
| ) | ||
|
|
||
| type ( | ||
| TesterApp struct { | ||
| Ctx context.Context | ||
| CancelFunc context.CancelFunc | ||
| CatalystPipelineStrategy string | ||
| Lapi *api.Client | ||
| } | ||
|
|
||
| TesterOptions struct { | ||
| API *api.Client | ||
| CatalystPipelineStrategy string | ||
| } | ||
| ) | ||
|
|
||
| func (ta *TesterApp) CheckTaskProcessing(taskPollDuration time.Duration, processingTask api.Task) error { | ||
| startTime := time.Now() | ||
| for { | ||
| time.Sleep(taskPollDuration) | ||
|
|
||
| if err := ta.isCancelled(); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // we already sleep before the first check, so no need for strong consistency | ||
| task, err := ta.Lapi.GetTask(processingTask.ID, false) | ||
| if err != nil { | ||
| glog.Errorf("Error retrieving task id=%s err=%v", processingTask.ID, err) | ||
| return fmt.Errorf("error retrieving task id=%s: %w", processingTask.ID, err) | ||
| } | ||
| if task.Status.Phase == "completed" { | ||
| glog.Infof("Task success, taskId=%s", task.ID) | ||
| return nil | ||
| } | ||
| if task.Status.Phase != "pending" && task.Status.Phase != "running" && task.Status.Phase != "waiting" { | ||
| glog.Errorf("Error processing task, taskId=%s status=%s error=%v", task.ID, task.Status.Phase, task.Status.ErrorMessage) | ||
| return fmt.Errorf("error processing task, taskId=%s status=%s error=%v", task.ID, task.Status.Phase, task.Status.ErrorMessage) | ||
| } | ||
|
|
||
| glog.Infof("Waiting for task to be processed id=%s pollWait=%s elapsed=%s progressPct=%.1f%%", task.ID, taskPollDuration, time.Since(startTime), 100*task.Status.Progress) | ||
| } | ||
| } | ||
|
|
||
| func (ta *TesterApp) isCancelled() error { | ||
| select { | ||
| case <-ta.Ctx.Done(): | ||
| return context.Canceled | ||
| default: | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (ta *TesterApp) Cancel() { | ||
| ta.CancelFunc() | ||
| } | ||
|
|
||
| func (ta *TesterApp) Done() <-chan struct{} { | ||
| return ta.Ctx.Done() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
internal/app/transcodetester/continuous_transcode_tester.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package transcodetester | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/livepeer/stream-tester/internal/app/common" | ||
| ) | ||
|
|
||
| type ( | ||
| IContinuousTranscodeTester interface { | ||
| // Start test. Blocks until error. | ||
| Start(fileName string, transcodeBucketUrl string, testDuration, taskPollDuration, pauseBetweenTests time.Duration) error | ||
| } | ||
|
|
||
| continuousTranscodeTester struct { | ||
| ct common.IContinuousTester | ||
| opts common.TesterOptions | ||
| } | ||
| ) | ||
|
|
||
| func NewContinuousTranscodeTester(gctx context.Context, opts common.ContinuousTesterOptions) IContinuousTranscodeTester { | ||
| return &continuousTranscodeTester{ | ||
| ct: common.NewContinuousTester(gctx, opts, "transcode"), | ||
| opts: opts.TesterOptions, | ||
| } | ||
| } | ||
|
|
||
| func (ctt *continuousTranscodeTester) Start(fileName string, transcodeBucketUrl string, testDuration, taskPollDuration, pauseBetweenTests time.Duration) error { | ||
| start := func(ctx context.Context) error { | ||
| tt := NewTranscodeTester(ctx, ctt.opts) | ||
| return tt.Start(fileName, transcodeBucketUrl, taskPollDuration) | ||
| } | ||
| return ctt.ct.Start(start, testDuration, pauseBetweenTests) | ||
| } |
Oops, something went wrong.