EurekaContainerHealthService implements ContainerHealthService { @Override public Flux<ContainerHealthEvent> events(boolean snapshot) { Preconditions.checkArgument(!snapshot, "Snapshot state is generated by AggregatingContainerHealthService"); return healthStatuses; } @Inject  EurekaContainerHealthService(ReadOnlyJobOperations jobOperations, EurekaClient eurekaClient, TitusRuntime titusRuntime); @Activator void activate(); @PreDestroy void shutdown(); @Override String getName(); @Override Optional<ContainerHealthStatus> findHealthStatus(String taskId); @Override Flux<ContainerHealthEvent> events(boolean snapshot); static final String NAME; }
@Test public void testJobManagerUpdate() { StepVerifier.create(healthService.events(false)) .then(() -> jobManagerStub.moveTaskToState(taskId1, TaskState.Launched)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unknown)) .then(() -> { eurekaServer.register(newInstanceInfo(taskId1, InstanceStatus.UP)); jobManagerStub.moveTaskToState(taskId1, TaskState.Started); }) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> jobManagerStub.moveTaskToState(task1, TaskState.Finished)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Terminated)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testEurekaUpdate() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); StepVerifier.create(healthService.events(false)) .then(() -> registerAndRefresh(InstanceStatus.UP)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> registerAndRefresh(InstanceStatus.DOWN)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unhealthy)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testEurekaReRegistration() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); StepVerifier.create(healthService.events(false)) .then(() -> registerAndRefresh(InstanceStatus.UP)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> { eurekaServer.unregister(taskId1); eurekaServer.triggerCacheRefreshUpdate(); }) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unknown)) .then(() -> registerAndRefresh(InstanceStatus.UP)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testOutOfServiceJobWithRealStateUp() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); StepVerifier.create(healthService.events(false)) .then(() -> { jobManagerStub.changeJobEnabledStatus(job1, false); registerAndRefresh(InstanceStatus.OUT_OF_SERVICE); }) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testOutOfServiceJobWithRealStateDown() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); StepVerifier.create(healthService.events(false)) .then(() -> registerAndRefresh(InstanceStatus.DOWN)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unhealthy)) .then(() -> jobManagerStub.changeJobEnabledStatus(job1, false)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> registerAndRefresh(InstanceStatus.OUT_OF_SERVICE)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> jobManagerStub.changeJobEnabledStatus(job1, true)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unhealthy)) .then(() -> registerAndRefresh(InstanceStatus.UP)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .thenCancel() .verify(Duration.ofSeconds(5_000)); }
@Test public void testOutOfServiceJobWithRealStateNotRegistered() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); StepVerifier.create(healthService.events(false)) .then(eurekaServer::triggerCacheRefreshUpdate) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unknown)) .then(() -> jobManagerStub.changeJobEnabledStatus(job1, false)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Healthy)) .then(() -> jobManagerStub.changeJobEnabledStatus(job1, true)) .assertNext(event -> assertContainerHealthAndEvent(event, healthService.getHealthStatus(taskId1), taskId1, ContainerHealthState.Unknown)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testBadSubscriberIsolation() { jobManagerStub.moveTaskToState(taskId1, TaskState.Started); eurekaServer.register(newInstanceInfo(taskId1, InstanceStatus.UP)); TitusRxSubscriber<ContainerHealthEvent> subscriber1 = new TitusRxSubscriber<>(); healthService.events(false).subscribe(subscriber1); eurekaServer.triggerCacheRefreshUpdate(); Disposable subscription2 = healthService.events(false).subscribe( next -> { throw new RuntimeException("simulated error"); }, e -> { throw new RuntimeException("simulated error"); }, () -> { throw new RuntimeException("simulated error"); } ); registerAndRefresh(InstanceStatus.DOWN); assertThat(subscription2.isDisposed()).isTrue(); registerAndRefresh(InstanceStatus.UP); assertThat(subscriber1.isDisposed()).isFalse(); assertThat(subscriber1.getAllItems()).hasSize(3); }