This is an automated email from the ASF dual-hosted git repository. wilfreds pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push: new 039f28f7 [YUNIKORN-2010] Application metric incorrect (#667) 039f28f7 is described below commit 039f28f7268e7e78054f64bd876a30b317adc294 Author: Yu-Lin Chen <kh87...@gmail.com> AuthorDate: Fri Nov 3 11:58:09 2023 +1100 [YUNIKORN-2010] Application metric incorrect (#667) The web UI shows negative number of applications running in the application history. Root cause is an incorrect tracking of running applications in the metrics. * remove 'enter_Resuming' metrics changes * remove '*_Completing' metrics changes * use 'enter_Running' for metrics changes * enhance metrics trace and add more test cases * correctly update accepted application metric Closes: #667 Signed-off-by: Wilfred Spiegelenburg <wilfr...@apache.org> --- pkg/metrics/init.go | 9 +- pkg/metrics/metrics_collector.go | 2 +- pkg/metrics/queue.go | 46 +++++ pkg/metrics/scheduler.go | 20 +- pkg/scheduler/objects/application_state.go | 40 ++-- pkg/scheduler/objects/application_state_test.go | 242 ++++++++++++++++++++++++ 6 files changed, 337 insertions(+), 22 deletions(-) diff --git a/pkg/metrics/init.go b/pkg/metrics/init.go index e7cdeb56..7a4f5b4e 100644 --- a/pkg/metrics/init.go +++ b/pkg/metrics/init.go @@ -47,11 +47,16 @@ type Metrics struct { type CoreQueueMetrics interface { IncQueueApplicationsAccepted() + GetQueueApplicationsAccepted() (int, error) IncQueueApplicationsRejected() + GetQueueApplicationsRejected() (int, error) IncQueueApplicationsRunning() DecQueueApplicationsRunning() + GetQueueApplicationsRunning() (int, error) IncQueueApplicationsFailed() + GetQueueApplicationsFailed() (int, error) IncQueueApplicationsCompleted() + GetQueueApplicationsCompleted() (int, error) IncAllocatedContainer() IncReleasedContainer() AddReleasedContainers(value int) @@ -102,6 +107,7 @@ type CoreSchedulerMetrics interface { // Metrics Ops related to TotalApplicationsRejected IncTotalApplicationsRejected() AddTotalApplicationsRejected(value int) + GetTotalApplicationsRejected() (int, error) // Metrics Ops related to TotalApplicationsRunning IncTotalApplicationsRunning() @@ -109,7 +115,7 @@ type CoreSchedulerMetrics interface { DecTotalApplicationsRunning() SubTotalApplicationsRunning(value int) SetTotalApplicationsRunning(value int) - getTotalApplicationsRunning() (int, error) + GetTotalApplicationsRunning() (int, error) // Metrics Ops related to TotalApplicationsFailed IncTotalApplicationsFailed() @@ -120,6 +126,7 @@ type CoreSchedulerMetrics interface { DecTotalApplicationsCompleted() SubTotalApplicationsCompleted(value int) SetTotalApplicationsCompleted(value int) + GetTotalApplicationsCompleted() (int, error) // Metrics Ops related to ActiveNodes IncActiveNodes() diff --git a/pkg/metrics/metrics_collector.go b/pkg/metrics/metrics_collector.go index ee9651ed..12bcb13e 100644 --- a/pkg/metrics/metrics_collector.go +++ b/pkg/metrics/metrics_collector.go @@ -67,7 +67,7 @@ func (u *internalMetricsCollector) StartService() { func (u *internalMetricsCollector) store() { log.Log(log.Metrics).Debug("Adding current status to historical partition data") - totalAppsRunning, err := m.scheduler.getTotalApplicationsRunning() + totalAppsRunning, err := m.scheduler.GetTotalApplicationsRunning() if err != nil { log.Log(log.Metrics).Warn("Could not encode totalApplications metric.", zap.Error(err)) totalAppsRunning = -1 diff --git a/pkg/metrics/queue.go b/pkg/metrics/queue.go index 8e0dc70a..cb227cf4 100644 --- a/pkg/metrics/queue.go +++ b/pkg/metrics/queue.go @@ -20,6 +20,7 @@ package metrics import ( "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" "go.uber.org/zap" "github.com/apache/yunikorn-core/pkg/log" @@ -94,22 +95,67 @@ func (m *QueueMetrics) DecQueueApplicationsRunning() { m.appMetrics.With(prometheus.Labels{"state": "running"}).Dec() } +func (m *QueueMetrics) GetQueueApplicationsRunning() (int, error) { + metricDto := &dto.Metric{} + err := m.appMetrics.With(prometheus.Labels{"state": "running"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *QueueMetrics) IncQueueApplicationsAccepted() { m.appMetrics.With(prometheus.Labels{"state": "accepted"}).Inc() } +func (m *QueueMetrics) GetQueueApplicationsAccepted() (int, error) { + metricDto := &dto.Metric{} + err := m.appMetrics.With(prometheus.Labels{"state": "accepted"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *QueueMetrics) IncQueueApplicationsRejected() { m.appMetrics.With(prometheus.Labels{"state": "rejected"}).Inc() } +func (m *QueueMetrics) GetQueueApplicationsRejected() (int, error) { + metricDto := &dto.Metric{} + err := m.appMetrics.With(prometheus.Labels{"state": "rejected"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *QueueMetrics) IncQueueApplicationsFailed() { m.appMetrics.With(prometheus.Labels{"state": "failed"}).Inc() } +func (m *QueueMetrics) GetQueueApplicationsFailed() (int, error) { + metricDto := &dto.Metric{} + err := m.appMetrics.With(prometheus.Labels{"state": "failed"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *QueueMetrics) IncQueueApplicationsCompleted() { m.appMetrics.With(prometheus.Labels{"state": "completed"}).Inc() } +func (m *QueueMetrics) GetQueueApplicationsCompleted() (int, error) { + metricDto := &dto.Metric{} + err := m.appMetrics.With(prometheus.Labels{"state": "completed"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *QueueMetrics) IncAllocatedContainer() { m.containerMetrics.With(prometheus.Labels{"state": "allocated"}).Inc() } diff --git a/pkg/metrics/scheduler.go b/pkg/metrics/scheduler.go index b7721eae..0134c3c7 100644 --- a/pkg/metrics/scheduler.go +++ b/pkg/metrics/scheduler.go @@ -264,6 +264,15 @@ func (m *SchedulerMetrics) AddTotalApplicationsRejected(value int) { m.applicationSubmission.With(prometheus.Labels{"result": "rejected"}).Add(float64(value)) } +func (m *SchedulerMetrics) GetTotalApplicationsRejected() (int, error) { + metricDto := &dto.Metric{} + err := m.applicationSubmission.With(prometheus.Labels{"result": "rejected"}).Write(metricDto) + if err == nil { + return int(*metricDto.Counter.Value), nil + } + return -1, err +} + func (m *SchedulerMetrics) IncTotalApplicationsRunning() { m.application.With(prometheus.Labels{"state": "running"}).Inc() } @@ -284,7 +293,7 @@ func (m *SchedulerMetrics) SetTotalApplicationsRunning(value int) { m.application.With(prometheus.Labels{"state": "running"}).Set(float64(value)) } -func (m *SchedulerMetrics) getTotalApplicationsRunning() (int, error) { +func (m *SchedulerMetrics) GetTotalApplicationsRunning() (int, error) { metricDto := &dto.Metric{} err := m.application.With(prometheus.Labels{"state": "running"}).Write(metricDto) if err == nil { @@ -317,6 +326,15 @@ func (m *SchedulerMetrics) SetTotalApplicationsCompleted(value int) { m.application.With(prometheus.Labels{"state": "completed"}).Set(float64(value)) } +func (m *SchedulerMetrics) GetTotalApplicationsCompleted() (int, error) { + metricDto := &dto.Metric{} + err := m.application.With(prometheus.Labels{"state": "completed"}).Write(metricDto) + if err == nil { + return int(*metricDto.Gauge.Value), nil + } + return -1, err +} + func (m *SchedulerMetrics) IncActiveNodes() { m.node.With(prometheus.Labels{"state": "active"}).Inc() } diff --git a/pkg/scheduler/objects/application_state.go b/pkg/scheduler/objects/application_state.go index 35dbc6b3..925d80f4 100644 --- a/pkg/scheduler/objects/application_state.go +++ b/pkg/scheduler/objects/application_state.go @@ -169,29 +169,29 @@ func NewAppState() *fsm.FSM { fmt.Sprintf("enter_%s", Starting.String()): func(_ context.Context, event *fsm.Event) { app := event.Args[0].(*Application) //nolint:errcheck app.startTime = time.Now() - app.queue.incRunningApps(app.ApplicationID) app.setStateTimer(app.startTimeout, app.stateMachine.Current(), RunApplication) + app.queue.incRunningApps(app.ApplicationID) metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsRunning() metrics.GetSchedulerMetrics().IncTotalApplicationsRunning() }, - fmt.Sprintf("enter_%s", Resuming.String()): func(_ context.Context, event *fsm.Event) { - app := event.Args[0].(*Application) //nolint:errcheck - metrics.GetQueueMetrics(app.queuePath).DecQueueApplicationsRunning() - metrics.GetSchedulerMetrics().DecTotalApplicationsRunning() - }, - fmt.Sprintf("enter_%s", Completing.String()): func(_ context.Context, event *fsm.Event) { - app := event.Args[0].(*Application) //nolint:errcheck - if event.Src == Starting.String() { + fmt.Sprintf("leave_%s", Starting.String()): func(_ context.Context, event *fsm.Event) { + if event.Dst != Running.String() { + app := event.Args[0].(*Application) //nolint:errcheck app.queue.decRunningApps() metrics.GetQueueMetrics(app.queuePath).DecQueueApplicationsRunning() metrics.GetSchedulerMetrics().DecTotalApplicationsRunning() } + }, + fmt.Sprintf("enter_%s", Completing.String()): func(_ context.Context, event *fsm.Event) { + app := event.Args[0].(*Application) //nolint:errcheck app.setStateTimer(completingTimeout, app.stateMachine.Current(), CompleteApplication) }, fmt.Sprintf("leave_%s", New.String()): func(_ context.Context, event *fsm.Event) { - app := event.Args[0].(*Application) //nolint:errcheck - metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsAccepted() - metrics.GetSchedulerMetrics().IncTotalApplicationsAccepted() + if event.Dst != Rejected.String() { + app := event.Args[0].(*Application) //nolint:errcheck + metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsAccepted() + metrics.GetSchedulerMetrics().IncTotalApplicationsAccepted() + } }, fmt.Sprintf("enter_%s", Rejected.String()): func(_ context.Context, event *fsm.Event) { app := event.Args[0].(*Application) //nolint:errcheck @@ -205,19 +205,21 @@ func NewAppState() *fsm.FSM { app.rejectedMessage = event.Args[1].(string) //nolint:errcheck } }, + fmt.Sprintf("enter_%s", Running.String()): func(_ context.Context, event *fsm.Event) { + app := event.Args[0].(*Application) //nolint:errcheck + // account for going back into running state + if event.Src == Completing.String() { + app.queue.incRunningApps(app.ApplicationID) + metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsRunning() + metrics.GetSchedulerMetrics().IncTotalApplicationsRunning() + } + }, fmt.Sprintf("leave_%s", Running.String()): func(_ context.Context, event *fsm.Event) { app := event.Args[0].(*Application) //nolint:errcheck app.queue.decRunningApps() metrics.GetQueueMetrics(app.queuePath).DecQueueApplicationsRunning() metrics.GetSchedulerMetrics().DecTotalApplicationsRunning() }, - fmt.Sprintf("leave_%s", Completing.String()): func(_ context.Context, event *fsm.Event) { - app := event.Args[0].(*Application) //nolint:errcheck - // account for going back into running state - if event.Dst == Running.String() { - app.queue.incRunningApps(app.ApplicationID) - } - }, fmt.Sprintf("enter_%s", Completed.String()): func(_ context.Context, event *fsm.Event) { app := event.Args[0].(*Application) //nolint:errcheck metrics.GetSchedulerMetrics().IncTotalApplicationsCompleted() diff --git a/pkg/scheduler/objects/application_state_test.go b/pkg/scheduler/objects/application_state_test.go index 2f1bf564..7a07f808 100644 --- a/pkg/scheduler/objects/application_state_test.go +++ b/pkg/scheduler/objects/application_state_test.go @@ -27,6 +27,7 @@ import ( "github.com/apache/yunikorn-core/pkg/common" "github.com/apache/yunikorn-core/pkg/events" + "github.com/apache/yunikorn-core/pkg/metrics" "github.com/apache/yunikorn-scheduler-interface/lib/go/si" ) @@ -303,3 +304,244 @@ func TestAppStateTransitionEvents(t *testing.T) { isStateChangeEvent(t, appInfo, si.EventRecord_APP_EXPIRED, records[6]) isStateChangeEvent(t, appInfo, si.EventRecord_APP_RESUMING, records[7]) } + +// Test to verify metrics after applications state transition +// app-00001: New -> Resuming -> Accepted -> Starting -> Running -> Completing-> Completed +// app-00002: New -> Accepted -> Starting -> Completing -> Running -> Failing-> Failed +// app-00003: New -> Accepted -> Starting -> Failing -> Failed +// app-00004: New -> Rejected +// Final metrics will be: 0 running, 3 accepted, 1 completed, 2 failed and 1 rejected applications +func TestAppStateTransitionMetrics(t *testing.T) { //nolint:funlen + queue := createQueue(t, "root.metrics") + metrics.GetSchedulerMetrics().Reset() + // app-00001: New -> Resuming -> Accepted --> Starting -> Running -> Completing-> Completed + app := newApplication("app-00001", "default", "root.metrics") + app.SetQueue(queue) + assertState(t, app, nil, New.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + // New -> Resuming + err := app.HandleApplicationEvent(ResumeApplication) + assertState(t, app, err, Resuming.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 0) + // Resuming -> Accepted + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Accepted.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 0) + // Accepted -> Starting + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Starting.String()) + assertTotalAppsRunningMetrics(t, 1) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 1) + assertQueueApplicationsRunningMetrics(t, app, 1) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 0) + // Starting -> Running + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Running.String()) + assertTotalAppsRunningMetrics(t, 1) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 1) + assertQueueApplicationsRunningMetrics(t, app, 1) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 0) + // Running -> Completing + err = app.HandleApplicationEvent(CompleteApplication) + assertState(t, app, err, Completing.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 0) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 0) + // Completing -> Completed + err = app.HandleApplicationEvent(CompleteApplication) + assertState(t, app, err, Completed.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 1) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 1) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 0) + assertQueueApplicationsCompletedMetrics(t, app, 1) + + // app-00002: New -> Accepted -> Starting -> Completing -> Running -> Failing-> Failed + app = newApplication("app-00002", "default", "root.metrics") + app.SetQueue(queue) + assertState(t, app, nil, New.String()) + // New -> Accepted + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Accepted.String()) + // Accepted -> Starting + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Starting.String()) + // Starting -> Completing + err = app.HandleApplicationEvent(CompleteApplication) + assertState(t, app, err, Completing.String()) + // Completing -> Running + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Running.String()) + // Running -> Failing + err = app.HandleApplicationEvent(FailApplication) + assertState(t, app, err, Failing.String()) + // Failing -> Failed + err = app.HandleApplicationEvent(FailApplication) + assertState(t, app, err, Failed.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 1) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 2) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 1) + assertQueueApplicationsCompletedMetrics(t, app, 1) + + // app-00003: New -> Accepted -> Starting -> Failing -> Failed + app = newApplication("app-00003", "default", "root.metrics") + app.SetQueue(queue) + assertState(t, app, nil, New.String()) + // New -> Accepted + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Accepted.String()) + // Accepted -> Starting + err = app.HandleApplicationEvent(RunApplication) + assertState(t, app, err, Starting.String()) + // Starting -> Failing + err = app.HandleApplicationEvent(FailApplication) + assertState(t, app, err, Failing.String()) + // Failing -> Failed + err = app.HandleApplicationEvent(FailApplication) + assertState(t, app, err, Failed.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 1) + assertTotalAppsRejectedMetrics(t, 0) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 3) + assertQueueApplicationsRejectedMetrics(t, app, 0) + assertQueueApplicationsFailedMetrics(t, app, 2) + assertQueueApplicationsCompletedMetrics(t, app, 1) + + // app-00004: New -> Rejected + app = newApplication("app-00004", "default", "root.metrics") + app.SetQueue(queue) + assertState(t, app, nil, New.String()) + // New -> Rejected + err = app.HandleApplicationEvent(RejectApplication) + assertState(t, app, err, Rejected.String()) + assertTotalAppsRunningMetrics(t, 0) + assertTotalAppsCompletedMetrics(t, 1) + assertTotalAppsRejectedMetrics(t, 1) + assertQueueRunningApps(t, app, 0) + assertQueueApplicationsRunningMetrics(t, app, 0) + assertQueueApplicationsAcceptedMetrics(t, app, 3) + assertQueueApplicationsRejectedMetrics(t, app, 1) + assertQueueApplicationsFailedMetrics(t, app, 2) + assertQueueApplicationsCompletedMetrics(t, app, 1) +} + +func assertState(t testing.TB, app *Application, err error, expected string) { + t.Helper() + assert.NilError(t, err, fmt.Sprintf("no error expected when change state to %v", expected)) + assert.Equal(t, app.CurrentState(), expected, "application not in expected state.") +} + +func assertTotalAppsRunningMetrics(t testing.TB, expected int) { + t.Helper() + totalAppsRunning, err := metrics.GetSchedulerMetrics().GetTotalApplicationsRunning() + assert.NilError(t, err, "no error expected when getting total running application count.") + assert.Equal(t, totalAppsRunning, expected, "total running application metrics is not as expected.") +} + +func assertTotalAppsCompletedMetrics(t testing.TB, expected int) { + t.Helper() + totalAppsCompleted, err := metrics.GetSchedulerMetrics().GetTotalApplicationsCompleted() + assert.NilError(t, err, "no error expected when getting total completed application count.") + assert.Equal(t, totalAppsCompleted, expected, "total completed application metrics is not as expected.") +} + +func assertTotalAppsRejectedMetrics(t testing.TB, expected int) { + t.Helper() + totalAppsRejected, err := metrics.GetSchedulerMetrics().GetTotalApplicationsRejected() + assert.NilError(t, err, "no error expected when getting total rejected application count.") + assert.Equal(t, totalAppsRejected, expected, "total rejected application metrics is not as expected.") +} + +func assertQueueRunningApps(t testing.TB, app *Application, expected int) { + t.Helper() + runningApps := app.queue.runningApps + assert.Equal(t, runningApps, uint64(expected), "total running application in queue is not as expected.") +} + +func assertQueueApplicationsAcceptedMetrics(t testing.TB, app *Application, expected int) { + t.Helper() + queueApplicationsAccepted, err := metrics.GetQueueMetrics(app.queuePath).GetQueueApplicationsAccepted() + assert.NilError(t, err, "no error expected when getting total accepted application count in queue.") + assert.Equal(t, queueApplicationsAccepted, expected, "total accepted application metrics in queue is not as expected.") +} + +func assertQueueApplicationsRejectedMetrics(t testing.TB, app *Application, expected int) { + t.Helper() + queueApplicationsRejected, err := metrics.GetQueueMetrics(app.queuePath).GetQueueApplicationsRejected() + assert.NilError(t, err, "no error expected when getting total rejected application count in queue.") + assert.Equal(t, queueApplicationsRejected, expected, "total rejected application metrics in queue is not as expected.") +} + +func assertQueueApplicationsRunningMetrics(t testing.TB, app *Application, expected int) { + t.Helper() + queueApplicationsRunning, err := metrics.GetQueueMetrics(app.queuePath).GetQueueApplicationsRunning() + assert.NilError(t, err, "no error expected when getting total running application count in queue.") + assert.Equal(t, queueApplicationsRunning, expected, "total running application metrics in queue is not as expected.") +} + +func assertQueueApplicationsFailedMetrics(t testing.TB, app *Application, expected int) { + t.Helper() + queueApplicationsFailed, err := metrics.GetQueueMetrics(app.queuePath).GetQueueApplicationsFailed() + assert.NilError(t, err, "no error expected when getting total failed application count in queue.") + assert.Equal(t, queueApplicationsFailed, expected, "total failed application metrics in queue is not as expected.") +} + +func assertQueueApplicationsCompletedMetrics(t testing.TB, app *Application, expected int) { + t.Helper() + queueApplicationsCompleted, err := metrics.GetQueueMetrics(app.queuePath).GetQueueApplicationsCompleted() + assert.NilError(t, err, "no error expected when getting total completed application count in queue.") + assert.Equal(t, queueApplicationsCompleted, expected, "total completed application metrics in queue is not as expected.") +} + +func createQueue(t *testing.T, queueName string) *Queue { + root, err := createRootQueue(nil) + assert.NilError(t, err, "failed to create queue: %v", err) + queue, err := createManagedQueue(root, queueName, false, map[string]string{"cpu": "10"}) + assert.NilError(t, err, "failed to create queue: %v", err) + return queue +} --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@yunikorn.apache.org For additional commands, e-mail: issues-h...@yunikorn.apache.org