[FLINK-33634] Add Conditions to Flink CRD's Status field #957
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,17 +18,22 @@ | |
| package org.apache.flink.kubernetes.operator.api.status; | ||
|
|
||
| import org.apache.flink.annotation.Experimental; | ||
| import org.apache.flink.api.common.JobStatus; | ||
| import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec; | ||
| import org.apache.flink.kubernetes.operator.api.utils.ConditionUtils; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonIgnoreProperties; | ||
| import io.fabric8.kubernetes.api.model.Condition; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Data; | ||
| import lombok.EqualsAndHashCode; | ||
| import lombok.NoArgsConstructor; | ||
| import lombok.ToString; | ||
| import lombok.experimental.SuperBuilder; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| /** Last observed status of the Flink deployment. */ | ||
|
|
@@ -55,4 +60,175 @@ public class FlinkDeploymentStatus extends CommonStatus<FlinkDeploymentSpec> { | |
|
|
||
| /** Information about the TaskManagers for the scale subresource. */ | ||
| private TaskManagerInfo taskManager; | ||
|
|
||
| /** Condition of the CR . */ | ||
| private List<Condition> conditions = new ArrayList<>(); | ||
|
|
||
| private String phase; | ||
|
|
||
| public List<Condition> getConditions() { | ||
| if (reconciliationStatus != null | ||
| && reconciliationStatus.deserializeLastReconciledSpec() != null | ||
| && reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) { | ||
|
There was a problem hiding this comment. We should avoid deserializing twice here, could we simply check the job status instead? There was a problem hiding this comment. If we derive the mode from Jobstatus.state, it looks like , there is a chance of misleading information in the condition status. For example, it is observed that, when an existing session cluster is updated with some information , for example with checkpoint interval changed to new value, we can see that state is updated with value as FINISHED as below status: So, if we derive the running state from JobStatus, by looking at the state value , we end up condition for above as Running false. But if we look at the jobManagerDeploymentStatus: READY, we are supposed to say that condition as Running true as it is session cluster, and it's says that session is ready to receive the request. |
||
| switch (jobManagerDeploymentStatus) { | ||
|
There was a problem hiding this comment. can we have a datastructure, keyed off the jobManagerDeploymentStatus, that we interrogate to get the inserts for the updateConditions. Maybe a map with the key of the status and the value of an object RunningCondition, that has 2 fields the boolean and the description. Something like : then the code just loops through the map updating conditions using the map values. Similar for jobstatus There was a problem hiding this comment. Wondering will that add any value?. Any way to build the Map , we have to call There was a problem hiding this comment. I think you have pre built the map now - with map.of - looks good - thanks. |
||
| case READY: | ||
|
There was a problem hiding this comment. instead of this switch can we just issue There was a problem hiding this comment. @davidradl this has been revisited and made it simple by removing switch. |
||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningTrue( | ||
| "JobManager is running and ready to receive REST API call", | ||
|
|
||
| "JobManager is running and ready to receive REST API call")); | ||
| break; | ||
| case MISSING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| "JobManager deployment not found", | ||
| "JobManager deployment not found")); | ||
| break; | ||
| case DEPLOYING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| "JobManager process is starting up", | ||
| "JobManager process is starting up")); | ||
|
There was a problem hiding this comment. I am curious why the message and reaon is the same . I would expect them to be different. In this case I would expect the reason to be "A New JVM deployment exists and is being created" - i.e. the word on the arrow in the UML. There was a problem hiding this comment. As per description here which says There was a problem hiding this comment. I see https://github.com/apache/flink-kubernetes-operator/pull/961/files, gives examples of camelcase single words reasons. We should follow that style for the cases that PR does not cover. There was a problem hiding this comment. Done a refactoring for the conditions. |
||
| break; | ||
| case DEPLOYED_NOT_READY: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| "JobManager is running but not ready yet to receive REST API calls", | ||
| "JobManager is running but not ready yet to receive REST API calls")); | ||
| break; | ||
| case ERROR: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| "Deployment in terminal error, requires spec change for reconciliation to continue", | ||
| "JobManager deployment failed")); | ||
| } | ||
| } else if (getJobStatus() != null && getJobStatus().getState() != null) { | ||
| switch (getJobStatus().getState()) { | ||
|
There was a problem hiding this comment. can this switch as-is be replaced with There was a problem hiding this comment. I was thinking exactly the same here @lajith2006 There was a problem hiding this comment. Yes, @gyfora and @davidradl this has been revisited and addressed by removing switch and made it simple. |
||
| case RECONCILING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.RECONCILING.name(), "Job is currently reconciling")); | ||
| break; | ||
| case CREATED: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.CREATED.name(), "Job is created")); | ||
| break; | ||
| case RUNNING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningTrue(JobStatus.RUNNING.name(), "Job is running")); | ||
| break; | ||
| case FAILING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.FAILING.name(), "Job has failed")); | ||
| break; | ||
| case RESTARTING: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.RESTARTING.name(), | ||
| "The job is currently undergoing a restarting")); | ||
|
There was a problem hiding this comment. nit: "The job is currently undergoing a restarting" -> "The job is currently restarting" |
||
| break; | ||
| case FAILED: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.FAILED.name(), | ||
| "The job has failed with a non-recoverable task failure")); | ||
| break; | ||
| case FINISHED: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.FINISHED.name(), | ||
| "Job's tasks have successfully finished")); | ||
| break; | ||
|
|
||
| case CANCELED: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.CANCELED.name(), "Job has been cancelled")); | ||
| break; | ||
| case SUSPENDED: | ||
| updateCondition( | ||
| conditions, | ||
| ConditionUtils.runningFalse( | ||
| JobStatus.SUSPENDED.name(), "The job has been suspended")); | ||
| break; | ||
| } | ||
| } | ||
| return conditions; | ||
| } | ||
|
|
||
| public String getPhase() { | ||
| if (reconciliationStatus != null | ||
| && reconciliationStatus.deserializeLastReconciledSpec() != null | ||
| && reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) { | ||
| switch (jobManagerDeploymentStatus) { | ||
| case READY: | ||
| phase = "Running"; | ||
| break; | ||
| case MISSING: | ||
| case ERROR: | ||
| case DEPLOYING: | ||
| phase = "Pending"; | ||
| break; | ||
| } | ||
| } else if (getJobStatus() != null && getJobStatus().getState() != null) { | ||
| switch (getJobStatus().getState()) { | ||
| case RECONCILING: | ||
| phase = "Pending"; | ||
|
There was a problem hiding this comment. why is this one not RECONCILING like the pattern the others follow others? I suggest a comment, also a constant is better then a an inline literal. There was a problem hiding this comment. Right, as mentioned here , As when FlinkDeployment applied in Application Mode, in the initial phase, as the JM brining up , job state would be in |
||
| break; | ||
| case CREATED: | ||
| phase = JobStatus.CREATED.name(); | ||
| break; | ||
| case RUNNING: | ||
| phase = JobStatus.RUNNING.name(); | ||
| break; | ||
| case FAILING: | ||
| phase = JobStatus.FAILING.name(); | ||
| break; | ||
| case RESTARTING: | ||
| phase = JobStatus.RESTARTING.name(); | ||
| break; | ||
| case FAILED: | ||
| phase = JobStatus.FAILED.name(); | ||
| break; | ||
| case FINISHED: | ||
| phase = JobStatus.FINISHED.name(); | ||
| break; | ||
| case CANCELED: | ||
| phase = JobStatus.CANCELED.name(); | ||
| break; | ||
| case SUSPENDED: | ||
| phase = JobStatus.SUSPENDED.name(); | ||
| break; | ||
| } | ||
| } | ||
| return phase; | ||
| } | ||
|
|
||
| private static void updateCondition(List<Condition> conditions, Condition condition) { | ||
| if (conditions.isEmpty()) { | ||
| conditions.add(condition); | ||
| return; | ||
| } | ||
| // If new condition is same as last condition, ignore | ||
| Condition existingCondition = conditions.get(conditions.size() - 1); | ||
| if (existingCondition.getType().equals(condition.getType()) | ||
| && existingCondition.getMessage().equals(condition.getMessage())) { | ||
| return; | ||
| } | ||
| conditions.add(condition); | ||
|
There was a problem hiding this comment. It's maybe just me, just don't get how this list is truncated eventually, in other words, how is it prevented from growing indefinitely? There was a problem hiding this comment. Thanks for review @csviri . Right now it's not preventing from growing indefinitely. As conditions reflects the different transitions state of Job/Deployment , thinking what could be the use case where can grow indefinitely?. There was a problem hiding this comment. I think there should only ever be a single condition of type There was a problem hiding this comment. Thanks @gyfora for review , If I read correctly , you meant that we need to have only condition in the list at right now as we have only one type There was a problem hiding this comment. @lajith2006 you say There was a problem hiding this comment. Yes @lajith2006 , there should be a single condition of type Running and lastTransitionTime when that changed There was a problem hiding this comment. Thanks for clarification @gyfora So just to be align with what you have mentioned , for an example , when we do an application cluster deployment, as Job cycles goes through CREATED>RUNNING initially conditions will have as below And then when the Job starts running at and then later at which means ideally we will not have an history in the condition when the Job goes through different job lifecycle , instead only single condition and lastTransitionTime will reflect when the status was changed. There was a problem hiding this comment. Yes, I think this makes sense for a |
||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.flink.kubernetes.operator.api.utils; | ||
|
|
||
| import io.fabric8.kubernetes.api.model.Condition; | ||
| import io.fabric8.kubernetes.api.model.ConditionBuilder; | ||
|
|
||
| import java.text.SimpleDateFormat; | ||
| import java.util.Date; | ||
|
|
||
| /** | ||
| * Creates a condition object with the specified parameters. | ||
|
There was a problem hiding this comment. nit: with the specified parameters -> whether this is running , the message and the reason. |
||
| * | ||
| * @return A condition object with the type, message, status, reason and timestamp. | ||
| */ | ||
| public class ConditionUtils { | ||
| public static Condition runningTrue(final String message, final String reason) { | ||
| return crCondition("Running", "True", message, reason); | ||
|
There was a problem hiding this comment. no need to pass "Running" as a parameter, we can hard code that in the method, nit: rename crCondition -> crRunningCondition There was a problem hiding this comment. I would like to keep method name as There was a problem hiding this comment. I suggest doing that refactor when we need to, in the spirit of keeping the code lean. There was a problem hiding this comment. @davidradl There were some refactoring done on this class. |
||
| } | ||
|
|
||
| public static Condition runningFalse(final String message, final String reason) { | ||
| return crCondition("Running", "False", message, reason); | ||
|
There was a problem hiding this comment. are there constants / enums for "False" and "True" we can use? There was a problem hiding this comment. I haven't see any existing enum/constants for "False" / "True , instead looks like used as string itself in wherever required in existing code. |
||
| } | ||
|
|
||
| private static Condition crCondition( | ||
| final String type, final String status, final String message, final String reason) { | ||
| return new ConditionBuilder() | ||
| .withType(type) | ||
| .withStatus(status) | ||
| .withMessage(message) | ||
| .withReason(reason) | ||
| .withLastTransitionTime( | ||
| new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(new Date())) | ||
| .build(); | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am curious which parts of the code refer to application mode and which are session cluster. It would be good to have some comments to detail this and maybe use the mode name in method name or variable names to make this more intuitive to read.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I will add comments in the respective code.