From 2f5eefe809b3b3be0dbc2d447270f23cdac68823 Mon Sep 17 00:00:00 2001 From: Jake McDermott Date: Thu, 10 May 2018 21:39:55 -0400 Subject: [PATCH 1/4] push initial events on partially complete job initialization --- .../features/output/index.controller.js | 18 +++++++----- awx/ui/client/features/output/index.js | 29 ++++++++++++------- .../client/features/output/render.service.js | 6 ++-- 3 files changed, 30 insertions(+), 23 deletions(-) diff --git a/awx/ui/client/features/output/index.controller.js b/awx/ui/client/features/output/index.controller.js index e84c93d708..b4d01c5a90 100644 --- a/awx/ui/client/features/output/index.controller.js +++ b/awx/ui/client/features/output/index.controller.js @@ -67,7 +67,6 @@ function init () { }); render.init({ - get: () => resource.model.get(`related.${resource.related}.results`), compile: html => $compile(html)($scope), isStreamActive: engine.isActive, }); @@ -95,21 +94,24 @@ function init () { } }); - $scope.$on(resource.ws.events, handleJobEvent); - $scope.$on(resource.ws.status, handleStatusEvent); - if (!status.state.running) { - next(); + return next(); } + + $scope.$on(resource.ws.events, (scope, data) => handleJobEvent(data)); + $scope.$on(resource.ws.status, (scope, data) => handleStatusEvent(data)); + + return resource.model + .get(`related.${resource.related}.results`) + .forEach(handleJobEvent); } -function handleStatusEvent (scope, data) { +function handleStatusEvent (data) { status.pushStatusEvent(data); } -function handleJobEvent (scope, data) { +function handleJobEvent (data) { engine.pushJobEvent(data); - status.pushJobEvent(data); } diff --git a/awx/ui/client/features/output/index.js b/awx/ui/client/features/output/index.js index 813d1c2af3..a8d635ed3a 100644 --- a/awx/ui/client/features/output/index.js +++ b/awx/ui/client/features/output/index.js @@ -70,13 +70,20 @@ function resolveResource ( return null; } - const params = { page_size: PAGE_SIZE, order_by: 'start_line' }; - const config = { pageCache: PAGE_CACHE, pageLimit: PAGE_LIMIT, params }; + const params = { + page_size: PAGE_SIZE, + order_by: 'start_line', + }; + + const config = { + params, + pageCache: PAGE_CACHE, + pageLimit: PAGE_LIMIT, + }; if (job_event_search) { // eslint-disable-line camelcase - const queryParams = qs.encodeQuerysetObject(qs.decodeArr(job_event_search)); - - Object.assign(config.params, queryParams); + const query = qs.encodeQuerysetObject(qs.decodeArr(job_event_search)); + Object.assign(config.params, query); } Wait('start'); @@ -89,7 +96,6 @@ function resolveResource ( } promises.push(model.extend('get', related, config)); - return Promise.all(promises); }) .then(([stats, model]) => ({ @@ -107,18 +113,19 @@ function resolveResource ( size: PAGE_SIZE, pageLimit: PAGE_LIMIT } - })) - .finally(() => Wait('stop')); + })); if (!handleErrors) { - return resourcePromise; + return resourcePromise + .finally(() => Wait('stop')); } return resourcePromise .catch(({ data, status }) => { - $state.go($state.current, $state.params, { reload: true }); qs.error(data, status); - }); + return $state.go($state.current, $state.params, { reload: true }); + }) + .finally(() => Wait('stop')); } function resolveWebSocketConnection ($stateParams, SocketService) { diff --git a/awx/ui/client/features/output/render.service.js b/awx/ui/client/features/output/render.service.js index 9981cac65b..08a3498fd2 100644 --- a/awx/ui/client/features/output/render.service.js +++ b/awx/ui/client/features/output/render.service.js @@ -30,11 +30,11 @@ const re = new RegExp(pattern); const hasAnsi = input => re.test(input); function JobRenderService ($q, $sce, $window) { - this.init = ({ compile, apply, isStreamActive }) => { + this.init = ({ compile, isStreamActive }) => { this.parent = null; this.record = {}; this.el = $(ELEMENT_TBODY); - this.hooks = { isStreamActive, compile, apply }; + this.hooks = { isStreamActive, compile }; }; this.sortByLineNumber = (a, b) => { @@ -239,8 +239,6 @@ function JobRenderService ($q, $sce, $window) { return list; }; - this.getEvents = () => this.hooks.get(); - this.insert = (events, insert) => { const result = this.transformEventGroup(events); const html = this.trustHtml(result.html); From f3343f780c70da9c711b54a3e6f338457432f940 Mon Sep 17 00:00:00 2001 From: Jake McDermott Date: Fri, 11 May 2018 16:42:55 -0400 Subject: [PATCH 2/4] always remove websocket listeners --- .../features/output/index.controller.js | 29 +++++++++++++------ 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/awx/ui/client/features/output/index.controller.js b/awx/ui/client/features/output/index.controller.js index b4d01c5a90..5ba68bc8e6 100644 --- a/awx/ui/client/features/output/index.controller.js +++ b/awx/ui/client/features/output/index.controller.js @@ -9,6 +9,7 @@ let engine; let status; let vm; +let listeners = []; function JobsIndexController ( _resource_, @@ -89,21 +90,32 @@ function init () { status.setJobStatus('running'); }, onStop () { + stopListening(); status.updateStats(); status.dispatch(); } }); if (!status.state.running) { - return next(); + next(); + return; } - $scope.$on(resource.ws.events, (scope, data) => handleJobEvent(data)); - $scope.$on(resource.ws.status, (scope, data) => handleStatusEvent(data)); - - return resource.model - .get(`related.${resource.related}.results`) + resource.model.get(`related.${resource.related}.results`) .forEach(handleJobEvent); + + startListening(); +} + +function stopListening () { + listeners.forEach(deregister => deregister()); + listeners = []; +} + +function startListening () { + stopListening(); + listeners.push($scope.$on(resource.ws.events, (scope, data) => handleJobEvent(data))); + listeners.push($scope.$on(resource.ws.status, (scope, data) => handleStatusEvent(data))); } function handleStatusEvent (data) { @@ -115,9 +127,8 @@ function handleJobEvent (data) { status.pushJobEvent(data); } -function devClear (pageMode) { - init(pageMode); - render.clear(); +function devClear () { + render.clear().then(() => init()); } function next () { From 665354c32e19b4df83daff960235a25fcf4bf359 Mon Sep 17 00:00:00 2001 From: Jake McDermott Date: Fri, 11 May 2018 19:27:45 -0400 Subject: [PATCH 3/4] add skip functionality to event replay tool --- .../management/commands/replay_job_events.py | 42 ++++++++++++------- 1 file changed, 26 insertions(+), 16 deletions(-) diff --git a/awx/main/management/commands/replay_job_events.py b/awx/main/management/commands/replay_job_events.py index 47ff723678..74bd9eb09f 100644 --- a/awx/main/management/commands/replay_job_events.py +++ b/awx/main/management/commands/replay_job_events.py @@ -95,7 +95,7 @@ class ReplayJobEvents(): raise RuntimeError("Job is of type {} and replay is not yet supported.".format(type(job))) sys.exit(1) - def run(self, job_id, speed=1.0, verbosity=0): + def run(self, job_id, speed=1.0, verbosity=0, skip=0): stats = { 'events_ontime': { 'total': 0, @@ -126,7 +126,10 @@ class ReplayJobEvents(): sys.exit(1) je_previous = None - for je_current in job_events: + for n, je_current in enumerate(job_events): + if n < skip: + continue + if not je_previous: stats['recording_start'] = je_current.created self.start(je_current.created) @@ -163,21 +166,25 @@ class ReplayJobEvents(): stats['events_total'] += 1 je_previous = je_current - - stats['replay_end'] = self.now() - stats['replay_duration'] = (stats['replay_end'] - stats['replay_start']).total_seconds() - stats['replay_start'] = stats['replay_start'].isoformat() - stats['replay_end'] = stats['replay_end'].isoformat() - stats['recording_end'] = je_current.created - stats['recording_duration'] = (stats['recording_end'] - stats['recording_start']).total_seconds() - stats['recording_start'] = stats['recording_start'].isoformat() - stats['recording_end'] = stats['recording_end'].isoformat() + if stats['events_total'] > 2: + stats['replay_end'] = self.now() + stats['replay_duration'] = (stats['replay_end'] - stats['replay_start']).total_seconds() + stats['replay_start'] = stats['replay_start'].isoformat() + stats['replay_end'] = stats['replay_end'].isoformat() + + stats['recording_end'] = je_current.created + stats['recording_duration'] = (stats['recording_end'] - stats['recording_start']).total_seconds() + stats['recording_start'] = stats['recording_start'].isoformat() + stats['recording_end'] = stats['recording_end'].isoformat() + + stats['events_ontime']['percentage'] = (stats['events_ontime']['total'] / float(stats['events_total'])) * 100.00 + stats['events_late']['percentage'] = (stats['events_late']['total'] / float(stats['events_total'])) * 100.00 + stats['events_distance_average'] = stats['events_distance_total'] / stats['events_total'] + stats['events_late']['lateness_average'] = stats['events_late']['lateness_total'] / stats['events_late']['total'] + else: + stats = {'events_total': stats['events_total']} - stats['events_ontime']['percentage'] = (stats['events_ontime']['total'] / float(stats['events_total'])) * 100.00 - stats['events_late']['percentage'] = (stats['events_late']['total'] / float(stats['events_total'])) * 100.00 - stats['events_distance_average'] = stats['events_distance_total'] / stats['events_total'] - stats['events_late']['lateness_average'] = stats['events_late']['lateness_total'] / stats['events_late']['total'] if verbosity >= 2: print(json.dumps(stats, indent=4, sort_keys=True)) @@ -191,11 +198,14 @@ class Command(BaseCommand): help='Id of the job to replay (job or adhoc)') parser.add_argument('--speed', dest='speed', type=int, metavar='s', help='Speedup factor.') + parser.add_argument('--skip', dest='skip', type=int, metavar='k', + help='Number of events to skip.') def handle(self, *args, **options): job_id = options.get('job_id') speed = options.get('speed') or 1 verbosity = options.get('verbosity') or 0 + skip = options.get('skip') or 0 replayer = ReplayJobEvents() - replayer.run(job_id, speed, verbosity) + replayer.run(job_id, speed, verbosity, skip) From 503668141bdfcfaa3b6bd326b123687c47f444df Mon Sep 17 00:00:00 2001 From: Jake McDermott Date: Fri, 11 May 2018 22:59:45 -0400 Subject: [PATCH 4/4] add procedure for attaching to running jobs --- .../client/features/output/engine.service.js | 10 +-- .../features/output/index.controller.js | 62 +++++++++++++++---- .../client/features/output/status.service.js | 22 +++++-- awx/ui/client/lib/models/Base.js | 8 +++ 4 files changed, 80 insertions(+), 22 deletions(-) diff --git a/awx/ui/client/features/output/engine.service.js b/awx/ui/client/features/output/engine.service.js index 1f74a90c59..2e6371520f 100644 --- a/awx/ui/client/features/output/engine.service.js +++ b/awx/ui/client/features/output/engine.service.js @@ -38,6 +38,12 @@ function JobEventEngine ($q) { }; }; + this.setMinLine = min => { + if (min > this.lines.min) { + this.lines.min = min; + } + }; + this.getBatchFactors = size => { const factors = [1]; @@ -140,10 +146,6 @@ function JobEventEngine ($q) { this.renderFrame = events => this.hooks.onEventFrame(events) .then(() => { - if (this.scroll.isLocked()) { - this.scroll.scrollToBottom(); - } - if (this.isEnding()) { const lastEvents = this.page.emptyBuffer(); diff --git a/awx/ui/client/features/output/index.controller.js b/awx/ui/client/features/output/index.controller.js index 5ba68bc8e6..5f5ba6b6eb 100644 --- a/awx/ui/client/features/output/index.controller.js +++ b/awx/ui/client/features/output/index.controller.js @@ -7,8 +7,10 @@ let resource; let scroll; let engine; let status; +let $http; let vm; +let streaming; let listeners = []; function JobsIndexController ( @@ -21,6 +23,7 @@ function JobsIndexController ( _$compile_, _$q_, _status_, + _$http_, ) { vm = this || {}; @@ -34,6 +37,7 @@ function JobsIndexController ( render = _render_; engine = _engine_; status = _status_; + $http = _$http_; // Development helper(s) vm.clear = devClear; @@ -96,15 +100,8 @@ function init () { } }); - if (!status.state.running) { - next(); - return; - } - - resource.model.get(`related.${resource.related}.results`) - .forEach(handleJobEvent); - - startListening(); + streaming = false; + return next().then(() => startListening()); } function stopListening () { @@ -123,12 +120,46 @@ function handleStatusEvent (data) { } function handleJobEvent (data) { - engine.pushJobEvent(data); - status.pushJobEvent(data); + streaming = streaming || attachToRunningJob(); + streaming.then(() => { + engine.pushJobEvent(data); + status.pushJobEvent(data); + }); } -function devClear () { - render.clear().then(() => init()); +function attachToRunningJob () { + const target = `${resource.model.get('url')}${resource.related}/`; + const params = { order_by: '-created', page_size: resource.page.size }; + + scroll.pause(); + + return render.clear() + .then(() => $http.get(target, { params })) + .then(res => { + const { results } = res.data; + + const minLine = 1 + Math.max(...results.map(event => event.end_line)); + const maxCount = Math.max(...results.map(event => event.counter)); + + const lastPage = resource.model.updateCount(maxCount); + + page.emptyCache(lastPage); + page.addPage(lastPage, [], true); + + engine.setMinLine(minLine); + + if (resource.model.page.current === lastPage) { + return $q.resolve(); + } + + return append(results); + }) + .then(() => { + scroll.setScrollPosition(scroll.getScrollHeight()); + scroll.resume(); + + return $q.resolve(); + }); } function next () { @@ -294,6 +325,10 @@ function toggleExpanded () { vm.expanded = !vm.expanded; } +function devClear () { + render.clear().then(() => init()); +} + // function showHostDetails (id) { // jobEvent.request('get', id) // .then(() => { @@ -344,6 +379,7 @@ JobsIndexController.$inject = [ '$compile', '$q', 'JobStatusService', + '$http', ]; module.exports = JobsIndexController; diff --git a/awx/ui/client/features/output/status.service.js b/awx/ui/client/features/output/status.service.js index 12e3fd2544..a4e60c6822 100644 --- a/awx/ui/client/features/output/status.service.js +++ b/awx/ui/client/features/output/status.service.js @@ -49,48 +49,60 @@ function JobStatusService (moment, message) { }; this.pushStatusEvent = data => { - const isJobEvent = (this.job === data.unified_job_id); - const isProjectEvent = (this.project && (this.project === data.project_id)); + const isJobStatusEvent = (this.job === data.unified_job_id); + const isProjectStatusEvent = (this.project && (this.project === data.project_id)); - if (isJobEvent) { + if (isJobStatusEvent) { this.setJobStatus(data.status); - } else if (isProjectEvent) { + this.dispatch(); + } else if (isProjectStatusEvent) { this.setProjectStatus(data.status); this.setProjectUpdateId(data.unified_job_id); + this.dispatch(); } }; this.pushJobEvent = data => { const isLatest = ((!this.counter) || (data.counter > this.counter)); + let changed = false; + if (!this.active && !(data.event === JOB_END)) { this.active = true; this.setJobStatus('running'); + changed = true; } if (isLatest) { this.counter = data.counter; this.latestTime = data.created; this.setElapsed(moment(data.created).diff(this.created, 'seconds')); + changed = true; } if (data.event === JOB_START) { this.setStarted(this.state.started || data.created); + changed = true; } if (data.event === PLAY_START) { this.state.counts.plays++; + changed = true; } if (data.event === TASK_START) { this.state.counts.tasks++; + changed = true; } if (data.event === JOB_END) { this.setStatsEvent(data); + changed = true; } - this.dispatch(); + if (changed) { + this.dispatch(); + } }; this.isExpectingStatsEvent = () => (this.jobType === 'job') || diff --git a/awx/ui/client/lib/models/Base.js b/awx/ui/client/lib/models/Base.js index 912d9a984c..a7cc321806 100644 --- a/awx/ui/client/lib/models/Base.js +++ b/awx/ui/client/lib/models/Base.js @@ -398,6 +398,13 @@ function extend (method, related, config = {}) { return Promise.reject(new Error(`No related property, ${related}, exists`)); } +function updateCount (count) { + this.page.count = count; + this.page.last = Math.ceil(count / this.page.size); + + return this.page.last; +} + function goToPage (config) { const params = config.params || {}; const { page } = config; @@ -693,6 +700,7 @@ function BaseModel (resource, settings) { this.extend = extend; this.copy = copy; this.getDependentResourceCounts = getDependentResourceCounts; + this.updateCount = updateCount; this.http = { get: httpGet.bind(this),