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 14 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 11 commits
Commits
Show all changes
14 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
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
60 changes: 60 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,18 +18,26 @@
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;
import java.util.Optional;

import static org.apache.flink.kubernetes.operator.api.utils.ConditionUtils.CONDITION_TYPE_RUNNING;

/** Last observed status of the Flink deployment. */
@Experimental
Expand All @@ -55,4 +63,56 @@ 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<>();

public List<Condition> getConditions() {
if (getJobStatus() != null) {
JobStatus jobStatus = getJobStatus().getState();
if (jobStatus == null) {
// Populate conditions for SessionMode deployment
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.SESSION_MODE_CONDITION.get(
jobManagerDeploymentStatus.name())));
} else if (jobStatus != null) {
Copy link

Choose a reason for hiding this comment

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

nit: } else if (jobStatus != null) { -> else {

// Populate conditions for ApplicationMode deployment
updateCondition(
conditions,
ConditionUtils.crCondition(
ConditionUtils.APPLICATION_MODE_CONDITION.get(jobStatus.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 think this entire code block should be part of the ConditionUtils probably encapsulated into a single:

ConditionUtils. updateCondition(status)
method.

Copy link
Author

Choose a reason for hiding this comment

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

Thanks @gyfora . I believe , If we have them in ConditionUtils block, we will not be able to utilise the jobManagerDeploymentStatus ConditionUtils. Please Correct me if I am wrong.

Copy link
Contributor

Choose a reason for hiding this comment

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

This could be implemented as a static method that operates on the status

Copy link

@davidradl davidradl May 15, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

This could be implemented as a static method that operates on the status

@gyfora Yes - we were thinking changing to something like

if (getJobStatus() != null) {
            JobStatus jobStatus = getJobStatus().getState();
            Condition conditionToAdd = null; 
            if (jobStatus == null) {
                // Populate conditions for SessionMode deployment
              conditionToAdd =
                                ConditionUtils.SESSION_MODE_CONDITION.get(
                                        jobManagerDeploymentStatus.name());
            } else {
                // Populate conditions for ApplicationMode deployment
                conditionToAdd =
                                ConditionUtils.APPLICATION_MODE_CONDITION.get(jobStatus.name());
            }
          conditions = ConditionsUtils.createListOfConditions(conditionToAdd);
}

Can you see a slicker way of doing this given we need jobManagerDeploymentStatus.name() for session and jobStatus.name() for application mode - which we know at this level. We could add a ternary if, I am not too worried about this.

Copy link
Contributor

Choose a reason for hiding this comment

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

I just wanted to factor out this condition logic from the FlinkDeploymentStatus class, it doesn't seem to belong there. I don't have an issue with he logic itself

Copy link
Author

Choose a reason for hiding this comment

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

Moved logic to ConditionUtils

return conditions;
}

private static void updateCondition(List<Condition> conditions, Condition newCondition) {
if (newCondition.getType().equals(CONDITION_TYPE_RUNNING)) {
Optional<Condition> existingCondition =
conditions.stream()
.filter(
c ->
c.getType().equals(CONDITION_TYPE_RUNNING)
&& c.getReason()
.equals(newCondition.getReason())
&& c.getMessage()
.equals(newCondition.getMessage()))
.findFirst();
// Until there is a condition change which reflects the latest state, no need to add
// condition to list.
if (existingCondition.isPresent()) {
return;
}
// Remove existing Condition with type running and then add a new condition that
Copy link

Choose a reason for hiding this comment

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

nit: the comment is not quite right I think.
I suggest this comment be
// Remove existing Condition with type running if not the same as the new condition

then add a new condition that reflects the current state. should be after the if

Copy link
Author

Choose a reason for hiding this comment

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

It's been refactored..

Copy link
Contributor

Choose a reason for hiding this comment

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

Honestly I don't get this whole logic here.
We should have a single condition at the moment of type RUNNING. So a at any point in time we either have a SingletonList or empty list.

Can we simply always update the list to List.of(newCondition) + update the transition timestamp if the value changed from true->false or vice versa.

Copy link
Contributor

Choose a reason for hiding this comment

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

@davidradl @lajith2006 am I missing something ?

Copy link

Choose a reason for hiding this comment

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

@gyfora I talked with @lajith2006. His code originally was wanting to handle more than one condition and manage the existing list to prevent 2 running conditions being in the list.

We agreed to keep it simple and amend this code as you suggest i.e. just add a new list with one running condition. We thought the transition time should also be updated if the message changes, as there are cases in the flow diagram in the Flip where the running condition remains false, but the message changes, in this case we think that we should update the transition time WDYT?

Copy link
Contributor

Choose a reason for hiding this comment

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

lastTransitionTime time should refer to the time when the status changes (true -> false or false->true) at least this is how it is used everywhere I know and everywhere I find online. Changed message alone should not update the timestamp as it would break the semantics and makes it impossible to tell when the job actually stopped running for example.

Copy link
Author

Choose a reason for hiding this comment

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

Thanks @gyfora , modified in such a way that lastTransitionTimeStamp gets updated when status changes from true> false or vice versa.

// reflects the current state.
conditions.removeIf(
c ->
c.getType().equals(CONDITION_TYPE_RUNNING)
&& !c.getMessage().equals(newCondition.getMessage())
Copy link

Choose a reason for hiding this comment

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

we have got here because the new condition is not the same as the existing condition. So we should just remove the existing condition without any conditions.

As-is the condition will not be removed if existing condition message or reason matches the new condition; which is not correct.

Copy link
Author

@lajith2006 lajith2006 May 14, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

So we should just remove the existing condition without any conditions.

Right its fine for now, But I think, if in later point of time, if we are adding new conditions other than Running, just removing the existing condition without of any check will cause an issue.

As-is the condition will not be removed if existing condition message or reason matches the new condition; which is not correct.

That's right, but for Running type , we don't have any such case were reason or message can be matched with existing condition.

&& !c.getReason().equals(newCondition.getReason()));
}
conditions.add(newCondition);
}
}
146 changes: 146 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,146 @@
/*
* 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 final String CONDITION_TYPE_RUNNING = "Running";

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(CONDITION_TYPE_RUNNING)
.withStatus("True")
.withReason("JobManagerReady")
.withMessage(
"JobManager is running and ready to receive REST API calls")
Copy link
Contributor

Choose a reason for hiding this comment

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

May be a better design to add this fields into the enum directly then simply get it from it.

Copy link
Author

Choose a reason for hiding this comment

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

Thanks, Sure, have addressed using enum .

.build(),
JobManagerDeploymentStatus.MISSING.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("Missing")
.withMessage("JobManager deployment not found")
.build(),
JobManagerDeploymentStatus.DEPLOYING.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("Deploying")
.withMessage("JobManager process is starting up")
.build(),
JobManagerDeploymentStatus.DEPLOYED_NOT_READY.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("DeployedNotReady")
.withMessage(
"JobManager is running but not ready yet to receive REST API calls")
.build(),
JobManagerDeploymentStatus.ERROR.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("Error")
.withMessage("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(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("Reconciling")
.withMessage("Job is currently reconciling")
.build(),
JobStatus.CREATED.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobCreated")
.withMessage("Job is created")
.build(),
JobStatus.RUNNING.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("True")
.withReason("JobRunning")
.withMessage("Job is running")
.build(),
JobStatus.FAILING.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobFailing")
.withMessage("Job has failed")
.build(),
JobStatus.RESTARTING.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobRestarting")
.withMessage("The job is currently restarting")
.build(),
JobStatus.FAILED.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobFailed")
.withMessage("The job has failed with a non-recoverable task failure")
.build(),
JobStatus.FINISHED.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobFinished")
.withMessage("Job's tasks have successfully finished")
.build(),
JobStatus.CANCELED.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobCancelled")
.withMessage("Job has been cancelled")
.build(),
JobStatus.SUSPENDED.name(),
new ConditionBuilder()
.withType(CONDITION_TYPE_RUNNING)
.withStatus("False")
.withReason("JobSuspended")
.withMessage("The job has been suspended")
.build());
}
Loading