@Test public void testOutlierResubmitWorks() throws InterruptedException { final String jobId = "test-job-1"; final int stage = 1; final int workerIdx = 0; final int workerNum = 1; final int numWorkers = 3; final int dropCount = 1; final int onNextCount = 9; final double dropPercent = dropCount * 100.0 / (dropCount + onNextCount); final List<GaugeMeasurement> outlierDropGauges = Arrays.asList( new GaugeMeasurement(MetricStringConstants.ON_NEXT_COUNT, onNextCount), new GaugeMeasurement(MetricStringConstants.DROP_COUNT, dropCount)); final List<GaugeMeasurement> zeroDropGauges = Arrays.asList( new GaugeMeasurement(MetricStringConstants.ON_NEXT_COUNT, onNextCount), new GaugeMeasurement(MetricStringConstants.DROP_COUNT, 0)); final MantisMasterClientApi mockMasterClientApi = mock(MantisMasterClientApi.class); final Map<Integer, WorkerAssignments> assignmentsMap = new HashMap<>(); Map<Integer, WorkerHost> hosts = new HashMap<>(); hosts.put(workerNum, new WorkerHost("localhost", workerIdx, Arrays.asList(31300), MantisJobState.Started, workerNum, 31301, -1)); hosts.put(workerNum + 1, new WorkerHost("localhost", workerIdx + 1, Arrays.asList(31302), MantisJobState.Started, workerNum, 31303, -1)); hosts.put(workerNum + 2, new WorkerHost("localhost", workerIdx + 2, Arrays.asList(31304), MantisJobState.Started, workerNum, 31305, -1)); assignmentsMap.put(stage, new WorkerAssignments(stage, numWorkers, hosts)); final CountDownLatch resubmitLatch = new CountDownLatch(1); final CountDownLatch autoScaleLatch = new CountDownLatch(1); when(mockMasterClientApi.schedulingChanges(jobId)).thenReturn(Observable.just(new JobSchedulingInfo(jobId, assignmentsMap))); when(mockMasterClientApi.resubmitJobWorker(anyString(), anyString(), anyInt(), anyString())).thenAnswer(new Answer<Observable<Boolean>>() { @Override public Observable<Boolean> answer(InvocationOnMock invocation) throws Throwable { final Object[] arguments = invocation.getArguments(); final String jobIdRecv = (String) arguments[0]; final String user = (String) arguments[1]; final int resubmittedWorkerNum = (Integer) arguments[2]; final Observable<Boolean> result = Observable.just(1).map(new Func1<Integer, Boolean>() { @Override public Boolean call(Integer integer) { logger.info("resubmitting worker {} of jobId {}", resubmittedWorkerNum, jobId); assertEquals(workerNum, resubmittedWorkerNum); assertEquals(user, "JobMaster"); assertEquals(jobId, jobIdRecv); resubmitLatch.countDown(); return true; } }); return result; } }); final AutoScaleMetricsConfig aggregationConfig = new AutoScaleMetricsConfig(); final WorkerMetricHandler workerMetricHandler = new WorkerMetricHandler(jobId, new Observer<JobAutoScaler.Event>() { @Override public void onCompleted() { logger.warn("onCompleted"); } @Override public void onError(Throwable e) { logger.warn("onError {}", e.getMessage(), e); } @Override public void onNext(JobAutoScaler.Event event) { logger.info("got auto scale event {}", event); JobAutoScaler.Event expected = new JobAutoScaler.Event(StageScalingPolicy.ScalingReason.DataDrop, stage, dropPercent / numWorkers, numWorkers, ""); assertEquals(expected, event); autoScaleLatch.countDown(); } }, mockMasterClientApi, aggregationConfig); final Observer<MetricData> metricDataObserver = workerMetricHandler.initAndGetMetricDataObserver(); final int minDataPointsForOutlierTrigger = 16; for (int i = 0; i <= minDataPointsForOutlierTrigger; i++) { metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx, workerNum, DATA_DROP_METRIC_GROUP, outlierDropGauges)); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx + 1, workerNum + 1, DATA_DROP_METRIC_GROUP, zeroDropGauges)); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx + 2, workerNum + 2, DATA_DROP_METRIC_GROUP, zeroDropGauges)); } assertTrue(resubmitLatch.await(30, TimeUnit.SECONDS)); assertTrue(autoScaleLatch.await(30 + 5, TimeUnit.SECONDS)); }
public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); Observer<MetricData> initAndGetMetricDataObserver(); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); Observer<MetricData> initAndGetMetricDataObserver();  }