KubeNotificationProcessor { private Mono<Void> updateTaskStatus(PodWrapper pod, Task task, TaskState newTaskState, Optional<TitusExecutorDetails> executorDetailsOpt, Optional<V1Node> node) { return ReactorExt.toMono(v3JobOperations.updateTask( task.getId(), currentTask -> Optional.of(updateTaskStatus(pod, newTaskState, executorDetailsOpt, node, currentTask, containerResultCodeResolver)), V3JobOperations.Trigger.Kube, "Kube pod notification", KUBE_CALL_METADATA )); } @Inject  KubeNotificationProcessor(JobManagerConfiguration configuration,
                                     DirectKubeApiServerIntegrator kubeApiServerIntegrator,
                                     KubeJobManagementReconciler kubeJobManagementReconciler,
                                     V3JobOperations v3JobOperations,
                                     ContainerResultCodeResolver containerResultCodeResolver); @Activator void enterActiveMode(); void shutdown();  }
@Test public void testUpdateTaskStatusVK() { when(containerResultCodeResolver.resolve(any(), any())).thenReturn(Optional.of("testUpdatedReasonCode")); V1Pod pod = new V1Pod() .metadata(new V1ObjectMeta() .name(TASK.getId()) ) .status(new V1PodStatus() .addContainerStatusesItem(new V1ContainerStatus() .containerID(TASK.getId()) .state(new V1ContainerState() .running(new V1ContainerStateRunning().startedAt(DateTime.now())) ) ) ); V1Node node = new V1Node() .metadata(new V1ObjectMeta() .annotations(CollectionsExt.asMap( TITUS_NODE_DOMAIN + "ami", "ami123", TITUS_NODE_DOMAIN + "stack", "myStack" )) ) .status(new V1NodeStatus() .addresses(Collections.singletonList( new V1NodeAddress().address("2.2.2.2").type(KubeUtil.TYPE_INTERNAL_IP) )) ); Task updatedTask = KubeNotificationProcessor.updateTaskStatus( new PodWrapper(pod), TaskState.Started, Optional.of(new TitusExecutorDetails(Collections.emptyMap(), new TitusExecutorDetails.NetworkConfiguration( true, "1.2.3.4", "", "1.2.3.4", "eniId123", "resourceId123" ))), Optional.of(node), TASK, containerResultCodeResolver ); Set<TaskState> pastStates = updatedTask.getStatusHistory().stream().map(ExecutableStatus::getState).collect(Collectors.toSet()); assertThat(pastStates).contains(TaskState.Accepted, TaskState.Launched, TaskState.StartInitiated); assertThat(updatedTask.getStatus().getReasonCode()).isEqualTo("testUpdatedReasonCode"); assertThat(updatedTask.getTaskContext()).containsEntry(TaskAttributes.TASK_ATTRIBUTES_AGENT_HOST, "2.2.2.2"); assertThat(updatedTask.getTaskContext()).containsEntry(TaskAttributes.TASK_ATTRIBUTES_CONTAINER_IP, "1.2.3.4"); assertThat(updatedTask.getTaskContext()).containsEntry(TaskAttributes.TASK_ATTRIBUTES_AGENT_AMI, "ami123"); assertThat(updatedTask.getTaskContext()).containsEntry(TaskAttributes.TASK_ATTRIBUTES_AGENT_STACK, "myStack"); }
@Test public void testUpdateTaskStatusKubelet() { when(containerResultCodeResolver.resolve(any(), any())).thenReturn(Optional.of("testUpdatedReasonCode")); V1Pod pod = new V1Pod() .metadata(new V1ObjectMeta() .name(TASK.getId()) ) .status(new V1PodStatus() .addContainerStatusesItem(new V1ContainerStatus() .containerID(TASK.getId()) .state(new V1ContainerState() .running(new V1ContainerStateRunning().startedAt(DateTime.now())) ) ) ); pod.getStatus().setPodIP("192.0.2.0"); V1Node node = new V1Node() .metadata(new V1ObjectMeta() .annotations(Collections.singletonMap( TITUS_NODE_DOMAIN + "ami", "ami123" )) ) .status(new V1NodeStatus() .addresses(Collections.singletonList( new V1NodeAddress().address("2.2.2.2").type(KubeUtil.TYPE_INTERNAL_IP) )) ); Task updatedTask = KubeNotificationProcessor.updateTaskStatus( new PodWrapper(pod), TaskState.Started, Optional.empty(), Optional.of(node), TASK, containerResultCodeResolver ); assertThat(updatedTask.getTaskContext()).containsEntry(TaskAttributes.TASK_ATTRIBUTES_CONTAINER_IP, "192.0.2.0"); }