@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"); }
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 )); }
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 )); } }
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); }
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(); }
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();  }