Skip to content

[FLINK-33634] Add Conditions to Flink CRD's Status field #957

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 17 commits into
base: main
Choose a base branch
from
Open

[FLINK-33634] Add Conditions to Flink CRD's Status field #957

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
afdf40b
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 12, 2025
1d84164
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 12, 2025
727bb4b
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 15, 2025
c66fd1f
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 19, 2025
c623f33
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 26, 2025
196c494
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 28, 2025
32294f4
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Mar 28, 2025
7b6d6a1
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 12, 2025
4f6679b
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 12, 2025
93be98a
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 14, 2025
9faab16
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 14, 2025
b40dcf3
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 16, 2025
70030db
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 16, 2025
9311031
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 May 18, 2025
9349ea4
Merge branch 'main' into 33634-status
lajith2006 Oct 7, 2025
00f5541
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Oct 9, 2025
b748ec7
[FLINK-33634] Add Conditions to Flink CRD's Status field
lajith2006 Oct 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 189 additions & 0 deletions .../src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand All @@ -55,4 +60,188 @@ 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
Copy link

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.

Copy link
Author

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.

&& reconciliationStatus.deserializeLastReconciledSpec() != null
&& reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should avoid deserializing twice here, could we simply check the job status instead?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure will use JobStatus.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:
clusterInfo:
jobManagerDeploymentStatus: READY
jobStatus:
checkpointInfo:
lastPeriodicCheckpointTimestamp: 0
savepointInfo:
lastPeriodicSavepointTimestamp: 0
savepointHistory: []
state: FINISHED

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.

// Populate conditions for SessionMode deployment
switch (jobManagerDeploymentStatus) {
Copy link

@davidradl davidradl Mar 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 :


Map<JobManagerDeploymentStatus,String> jobmanagerDeploymentStatusMap = new HashMap<String,String>() {{
    put(READY, new RunningCondition(true, "JobManager is running and ready to receive REST API calls")),
    ....
}};

then the code just loops through the map updating conditions using the map values. Similar for jobstatus

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wondering will that add any value?. Any way to build the Map , we have to call ConditionUtils to build the Condition, so rather than have a Prebuild Map , we can directly call ConditionUtils to build them right. Your thoughts?.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you have pre built the map now - with map.of - looks good - thanks.

case READY:
Copy link

@davidradl davidradl Apr 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of this switch can we just issue

updateCondition(
                           conditions,
                           ConditionUtils.crCondition(
                                   ConditionUtils.SESSION_MODE_CONDITION.get(
                                           jobManagerDeploymentStatus.name())));

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@davidradl this has been revisited and made it simple by removing switch.

updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
JobManagerDeploymentStatus.READY.name())));
break;
case MISSING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
JobManagerDeploymentStatus.MISSING.name())));
break;
case DEPLOYING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
JobManagerDeploymentStatus.DEPLOYING.name())));
break;
case DEPLOYED_NOT_READY:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
JobManagerDeploymentStatus.DEPLOYED_NOT_READY.name())));
break;
case ERROR:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
JobManagerDeploymentStatus.ERROR.name())));
}
} else if (getJobStatus() != null && getJobStatus().getState() != null) {
// Populate conditions for ApplicationMode deployment
switch (getJobStatus().getState()) {
Copy link

@davidradl davidradl Apr 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can this switch as-is be replaced with

 updateCondition(
                            conditions,
                            ConditionUtils.crCondition(
                                    ConditionUtils.APPLICATION_MODE_CONDITION.get(
                                            getJobStatus().getState().name())));

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was thinking exactly the same here @lajith2006

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, @gyfora and @davidradl this has been revisited and addressed by removing switch and made it simple.

case RECONCILING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.RECONCILING.name())));
break;
case CREATED:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.CREATED.name())));
break;
case RUNNING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.RUNNING.name())));
break;
case FAILING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.FAILING.name())));
break;
case RESTARTING:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.RESTARTING.name())));
break;
case FAILED:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.FAILED.name())));
break;
case FINISHED:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.FINISHED.name())));
break;

case CANCELED:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.CANCELED.name())));
break;
case SUSPENDED:
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(
JobStatus.SUSPENDED.name())));
break;
}
}
return conditions;
}

public String getPhase() {
if (reconciliationStatus != null
&& reconciliationStatus.deserializeLastReconciledSpec() != null
&& reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {
// populate phase for SessionMode deployment
switch (jobManagerDeploymentStatus) {
case READY:
phase = "Running";
break;
case MISSING:
case DEPLOYING:
phase = "Pending";
break;
case ERROR:
phase = "Failed";
break;
}
} else if (getJobStatus() != null && getJobStatus().getState() != null) {
// populate phase for ApplicationMode deployment
switch (getJobStatus().getState()) {
case RECONCILING:
phase = "Pending";
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, as mentioned here , phase which is intended to keep the current status of FlinkDeployment especially useful in Openshift environment, as Openshift UI can render the value from status.phase and populate the current status of deployment.

As when FlinkDeployment applied in Application Mode, in the initial phase, as the JM brining up , job state would be in RECONCILING state, so status.phase is kept as "Pending".

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);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there should only ever be a single condition of type Running in the list that we return. Since we only have a single condition type right now, then the list should only have a single element. The latestTransition timestamp needs to represent when running changed from true->false or false->true. We can however keep updating the message if we want.

Copy link
Author

@lajith2006 lajith2006 Apr 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 Running instead of multiple conditions of same type Running in the list as currently this PR having. And the lastTransitionTime in the condition must represent when the Running type changed its status from true>false or false > true. Is that correct?.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@lajith2006 you say the lastTransitionTime in the condition must represent when the Running type changed its status from true>false or false > true. Is that correct?. So I am curious what happens in the history if we change the reason Text? Can you check that if we change the reason text and not the running flag, we see all of the entries in the history if all the historical conditions have the same lastTransitionTime with changing reasons.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes @lajith2006 , there should be a single condition of type Running and lastTransitionTime when that changed

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

status
  conditions:
    - type: Running
      status: "False"
      reason: Job is created
      message: "Job is newly created, no task has started to run"
      lastTransitionTime: 2025-04-030T06:17:08Z

And then when the Job starts running at 2025-04-030T06:19:01Z , condition will have as below

status
  conditions:
    - type: Running
      status: "True"
      reason: Job is running
      message: "Job is running"
      lastTransitionTime: 2025-04-030T06:19:01Z

and then later at 2025-04-030T07:00:01Z. when job gets finished , condition will have as below.

status
  conditions:
    - type: Running
      status: "False"
      reason: Job's tasks have successfully finished
      message: Job's tasks have successfully finished
      lastTransitionTime: 2025-04-030T07:00:01Z

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.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I think this makes sense for a Running condition

}
}
143 changes: 143 additions & 0 deletions ...ator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
/*
* 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 org.apache.flink.api.common.JobStatus;
import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;

import io.fabric8.kubernetes.api.model.Condition;
import io.fabric8.kubernetes.api.model.ConditionBuilder;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

/** Creates a condition object with the type, status, message and reason. */
public class ConditionUtils {
public static Condition crCondition(Condition condition) {
return new ConditionBuilder(condition)
.withLastTransitionTime(
new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(new Date()))
.build();
}

public static final Map<String, Condition> SESSION_MODE_CONDITION =
Map.of(
JobManagerDeploymentStatus.READY.name(),
new ConditionBuilder()
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: you could refactor to a method and pass JobManagerDeploymentConditionStatus.READY, as the rest of the condition builder is the same apart from the status .
So

JobManagerDeploymentStatus.READY.name(),
createRunningConditionBuilder(JobManagerDeploymentConditionStatus.READY),
 JobManagerDeploymentStatus.MISSING.name(),
createRunningConditionBuilder(JobManagerDeploymentConditionStatus.MISSING),
...

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It has been moved to method and reason and messages are getting from enum.

.withType("Running")
.withStatus("True")
.withMessage("Ready")
.withReason("JobManager is running and ready to receive REST API calls")
Copy link

@davidradl davidradl Apr 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isn't the reason - how it got to this state, not a description of the state itself?

https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-conditions

it says:
reason | Machine-readable, UpperCamelCase text indicating the reason for the condition's last transition.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@davidradl this has been updated and reason as is made as one-word, CamelCase representation of current status.

.build(),
JobManagerDeploymentStatus.MISSING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("Missing")
.withReason("JobManager deployment not found")
.build(),
JobManagerDeploymentStatus.DEPLOYING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("Deploying")
.withReason("JobManager process is starting up")
.build(),
JobManagerDeploymentStatus.DEPLOYED_NOT_READY.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("DeployedNotReady")
.withReason(
"JobManager is running but not ready yet to receive REST API calls")
.build(),
JobManagerDeploymentStatus.ERROR.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("Error")
.withReason("JobManager deployment failed")
.build());

public static final Map<String, Condition> APPLICATION_MODE_CONDITION =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To me this entire mapping feels like that it should be replaced with a method,

return new ConditionBuilder()
                            .withType(CONDITION_TYPE_RUNNING)
                            .withStatus(status == RUNNING ? "True" : "False")
                            .withReason(status.name())
                            .withMessage("Job state " + status.name)
                            .build()

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, I have a them in method. As reason is expected to be in camelcase, I have added a method to make it as camecase from status.name().

Map.of(
JobStatus.RECONCILING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("Reconciling")
.withReason("Job is currently reconciling")
.build(),
JobStatus.CREATED.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobCreated")
.withReason("Job is created")
.build(),
JobStatus.RUNNING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("True")
.withMessage("JobRunning")
.withReason("Job is running")
.build(),
JobStatus.FAILING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobFailing")
.withReason("Job has failed")
.build(),
JobStatus.RESTARTING.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobRestarting")
.withReason("The job is currently restarting")
.build(),
JobStatus.FAILED.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobFailed")
.withReason("The job has failed with a non-recoverable task failure")
.build(),
JobStatus.FINISHED.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobFinished")
.withReason("Job's tasks have successfully finished")
.build(),
JobStatus.CANCELED.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobCancelled")
.withReason("Job has been cancelled")
.build(),
JobStatus.SUSPENDED.name(),
new ConditionBuilder()
.withType("Running")
.withStatus("False")
.withMessage("JobSuspended")
.withReason("The job has been suspended")
.build());
}
Loading