ExecutorHealthChecker { @VisibleForTesting void checkExecutorHealth() { final Map<Optional<Executor>, List<ExecutableFlow>> exFlowMap = getFlowToExecutorMap(); for (final Map.Entry<Optional<Executor>, List<ExecutableFlow>> entry : exFlowMap.entrySet()) { final Optional<Executor> executorOption = entry.getKey(); if (!executorOption.isPresent()) { final String finalizeReason = "Executor id of this execution doesn't exist."; finalizeFlows(entry.getValue(), finalizeReason); continue; } final Executor executor = executorOption.get(); Optional<ExecutorManagerException> healthcheckException = Optional.empty(); Map<String, Object> results = null; try { results = this.apiGateway .callWithExecutionId(executor.getHost(), executor.getPort(), ConnectorParams.PING_ACTION, null, null); } catch (final ExecutorManagerException e) { healthcheckException = Optional.of(e); } catch (final RuntimeException re) { logger.error("Unexepected exception while reaching executor - " + executorDetailString(executor), re); } if (!healthcheckException.isPresent()) { if (results == null || results.containsKey(ConnectorParams.RESPONSE_ERROR) || !results .containsKey(ConnectorParams.STATUS_PARAM) || !results.get(ConnectorParams.STATUS_PARAM) .equals(ConnectorParams.RESPONSE_ALIVE)) { healthcheckException = Optional.of( new ExecutorManagerException("Status of executor - " + executorDetailString(executor) + " is not alive.")); } } if (healthcheckException.isPresent()){ try { handleExecutorNotAliveCase(executor, entry.getValue(), healthcheckException.get()); } catch (RuntimeException re) { logger.error("Unchecked exception during failure handling for executor - " + executorDetailString(executor), re); } } else { if (this.executorFailureCount.containsKey(executor.getId())) { this.executorFailureCount.put(executor.getId(), 0); } } } } @Inject  ExecutorHealthChecker(final Props azkProps, final ExecutorLoader executorLoader,
      final ExecutorApiGateway apiGateway, final AlerterHolder alerterHolder); void start(); void shutdown(); void checkExecutorHealthQuietly();  }
@Test public void checkExecutorHealthAlive() throws Exception { this.activeFlows.put(EXECUTION_ID_11, new Pair<>( new ExecutionReference(EXECUTION_ID_11, this.executor1), this.flow1)); when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null)).thenReturn(ImmutableMap.of(ConnectorParams .STATUS_PARAM, ConnectorParams.RESPONSE_ALIVE)); this.executorHealthChecker.checkExecutorHealth(); assertThat(this.flow1.getStatus()).isEqualTo(Status.RUNNING); verifyZeroInteractions(this.alerterHolder); }
@Test public void checkExecutorHealthExecutorIdRemoved() throws Exception { this.activeFlows.put(EXECUTION_ID_11, new Pair<>( new ExecutionReference(EXECUTION_ID_11, null), this.flow1)); when(this.loader.fetchExecutableFlow(EXECUTION_ID_11)).thenReturn(this.flow1); this.executorHealthChecker.checkExecutorHealth(); verify(this.loader).updateExecutableFlow(this.flow1); assertThat(this.flow1.getStatus()).isEqualTo(Status.FAILED); }
@Test public void checkExecutorHealthConsecutiveFailures() throws Exception { this.activeFlows.put(EXECUTION_ID_11, new Pair<>( new ExecutionReference(EXECUTION_ID_11, this.executor1), this.flow1)); this.executorHealthChecker.checkExecutorHealth(); verify(this.apiGateway).callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null); verifyZeroInteractions(this.alerterHolder); when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null)).thenReturn(ImmutableMap.of(ConnectorParams .STATUS_PARAM, ConnectorParams.RESPONSE_ALIVE)); this.executorHealthChecker.checkExecutorHealth(); verifyZeroInteractions(this.alerterHolder); when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null)).thenReturn(null); this.executorHealthChecker.checkExecutorHealth(); verifyZeroInteractions(this.alerterHolder); when(this.loader.fetchExecutableFlow(flow1.getExecutionId())).thenReturn(flow1); this.executorHealthChecker.checkExecutorHealth(); verify((this.alerterHolder).get("email")) .alertOnFailedExecutorHealthCheck(eq(this.executor1), eq(Arrays.asList(this.flow1)), any(ExecutorManagerException.class), eq(Arrays.asList(AZ_ADMIN_ALERT_EMAIL.split(",")))); assertThat(this.flow1.getStatus()).isEqualTo(Status.FAILED); String expectedReason = "Executor was unreachable, executor-id: 1, executor-host: localhost, " + "executor-port: 12345"; verify((this.alerterHolder).get("email")).alertOnError(eq(flow1), eq(expectedReason)); }
@Test public void testFailureDuringExecutorPing() throws Exception { this.activeFlows.put(EXECUTION_ID_11, new Pair<>( new ExecutionReference(EXECUTION_ID_11, this.executor1), this.flow1)); this.activeFlows.put(EXECUTION_ID_12, new Pair<>( new ExecutionReference(EXECUTION_ID_12, this.executor2), this.flow2)); when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null)).thenThrow(new RuntimeException("test exception")); when(this.apiGateway.callWithExecutionId(this.executor2.getHost(), this.executor2.getPort(), ConnectorParams.PING_ACTION, null, null)).thenThrow(new RuntimeException("test exception")); this.executorHealthChecker.checkExecutorHealth(); verify(this.apiGateway).callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null); verify(this.apiGateway).callWithExecutionId(this.executor2.getHost(), this.executor2.getPort(), ConnectorParams.PING_ACTION, null, null); verifyZeroInteractions(this.alerterHolder); }
@Test public void testFailureDuringAlerting() throws Exception { this.activeFlows.clear(); this.activeFlows.put(EXECUTION_ID_11, new Pair<>( new ExecutionReference(EXECUTION_ID_11, this.executor1), this.flow1)); ExecutorManagerException healthcheckException = new ExecutorManagerException("test exception"); when(this.apiGateway.callWithExecutionId( this.executor1.getHost(), this.executor1.getPort(), ConnectorParams.PING_ACTION, null, null)) .thenThrow(healthcheckException); doThrow(new RuntimeException("test runtime exception")) .when(this.mailAlerter) .alertOnFailedExecutorHealthCheck( this.executor1, Arrays.asList(this.flow1), healthcheckException, Arrays.asList(AZ_ADMIN_ALERT_EMAIL.split(","))); when(this.loader.fetchExecutableFlow(EXECUTION_ID_11)).thenReturn(this.flow1); for (int failureCount = 0; failureCount < props.getInt(ConfigurationKeys.AZKABAN_EXECUTOR_MAX_FAILURE_COUNT); failureCount++) { this.executorHealthChecker.checkExecutorHealth(); } verify(this.loader).updateExecutableFlow(this.flow1); assertThat(this.flow1.getStatus()).isEqualTo(Status.FAILED); }