Dennis-Mircea commented on code in PR #1232:
URL: 
https://github.com/apache/flink-kubernetes-operator/pull/1232#discussion_r4398999682


##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -1243,6 +1247,74 @@ private static Configuration 
getOperatorRestConfig(Configuration origConfig)
         return conf;
     }
 
+    @VisibleForTesting
+    protected void forceDeleteStuckPods(Map<String, String> labels, Deployment 
deployment) {
+        var now = Instant.now();
+        var stuckPods =
+                kubernetesClient
+                        .pods()
+                        .inNamespace(deployment.getMetadata().getNamespace())
+                        .withLabels(labels)
+                        .list()
+                        .getItems()
+                        .stream()
+                        .filter(pod -> isPastDeletionDeadline(pod, now))
+                        .collect(Collectors.toList());
+        stuckPods.forEach(pod -> 
kubernetesClient.resource(pod).withGracePeriod(0).delete());
+
+        if (!stuckPods.isEmpty()) {
+            EventUtils.createOrUpdateEventWithInterval(
+                    kubernetesClient,
+                    deployment,
+                    EventRecorder.Type.Warning,
+                    FORCE_DELETED_EVENT_KEY,
+                    String.format(
+                            "Force deleted %d pod(s) that did not terminate in 
time "
+                                    + "(likely their node became unreachable), 
blocking cluster resource cleanup.",
+                            stuckPods.size()),
+                    EventRecorder.Component.JobManagerDeployment,
+                    event -> {},
+                    FORCE_DELETED_EVENT_KEY,
+                    null);
+        }
+    }
+
+    // The deletion timestamp of a pod already includes its termination grace 
period
+    private static boolean isPastDeletionDeadline(Pod pod, Instant now) {
+        var deletionTimestamp = pod.getMetadata().getDeletionTimestamp();
+        return deletionTimestamp != null && 
now.isAfter(Instant.parse(deletionTimestamp));
+    }
+
+    protected Duration deleteDeploymentBlockingWithForceDelete(
+            String name,
+            Resource<Deployment> deployment,
+            DeletionPropagation propagation,
+            Duration timeout,
+            Map<String, String> podLabels,
+            Configuration conf) {
+        if (!conf.get(
+                KubernetesOperatorConfigOptions
+                        .OPERATOR_RESOURCE_CLEANUP_FORCE_DELETE_STUCK_PODS)) {

Review Comment:
   Should this property be promoted to operator level? I'm asking this because, 
the other 2 properties that are used in this flow 
(`kubernetes.operator.resource.cleanup.timeout` and 
`kubernetes.operator.resource.deletion.propagation`) are at the operator level, 
not at the spec level.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -178,6 +178,19 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "The timeout for the resource clean up to wait for 
flink to shutdown cluster.");
 
+    @Documentation.Section(SECTION_SYSTEM_RECONCILE)
+    public static final ConfigOption<Boolean> 
OPERATOR_RESOURCE_CLEANUP_FORCE_DELETE_STUCK_PODS =
+            operatorConfig("resource.cleanup.force-delete-stuck-pods.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to force delete (grace period 0) the 
JobManager/TaskManager pods that are "
+                                    + "still Terminating past their deletion 
deadline (e.g. stuck due to their node "
+                                    + "becoming unreachable) while waiting for 
the cluster resources to be deleted. "
+                                    + "Stuck pods are force deleted once, 
after half of resource.cleanup.timeout has "

Review Comment:
   In the default propagation mode, which is `FOREGROUND`, the timeout is not 
actually half of `resource.cleanup.timeout`, but it is half of the remaining 
`resource.cleanup.timeout` after deducting the time spent on scaling the JM 
deployment pods to 0.
   
   Just wanted to mention this. The description of this property is already 
pretty big, and I don't know if it is worth making it even bigger.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -1243,6 +1247,74 @@ private static Configuration 
getOperatorRestConfig(Configuration origConfig)
         return conf;
     }
 
+    @VisibleForTesting
+    protected void forceDeleteStuckPods(Map<String, String> labels, Deployment 
deployment) {
+        var now = Instant.now();
+        var stuckPods =
+                kubernetesClient
+                        .pods()
+                        .inNamespace(deployment.getMetadata().getNamespace())
+                        .withLabels(labels)
+                        .list()
+                        .getItems()
+                        .stream()
+                        .filter(pod -> isPastDeletionDeadline(pod, now))
+                        .collect(Collectors.toList());
+        stuckPods.forEach(pod -> 
kubernetesClient.resource(pod).withGracePeriod(0).delete());

Review Comment:
   From what I checked, the `--grace-period=0` only skips the kubelet 
confirmation, which covers the unreachable node case, but what about the 
finalizers that can still be in place, because the grace period feature deletes 
the pods that have `metadata.finalizers` empty. From what I see this is a rare 
situation, because this is outside of JOSDK scope, as it implies to have an 
external component that adds additional finalizers on the resource. So, 
actually there can still be an edge case where some pods may still be alive and 
not deleted right away. I don't know if it makes sense to tackle this scenario.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -1243,6 +1247,74 @@ private static Configuration 
getOperatorRestConfig(Configuration origConfig)
         return conf;
     }
 
+    @VisibleForTesting
+    protected void forceDeleteStuckPods(Map<String, String> labels, Deployment 
deployment) {
+        var now = Instant.now();
+        var stuckPods =
+                kubernetesClient
+                        .pods()
+                        .inNamespace(deployment.getMetadata().getNamespace())
+                        .withLabels(labels)
+                        .list()
+                        .getItems()
+                        .stream()
+                        .filter(pod -> isPastDeletionDeadline(pod, now))
+                        .collect(Collectors.toList());
+        stuckPods.forEach(pod -> 
kubernetesClient.resource(pod).withGracePeriod(0).delete());
+
+        if (!stuckPods.isEmpty()) {
+            EventUtils.createOrUpdateEventWithInterval(
+                    kubernetesClient,
+                    deployment,
+                    EventRecorder.Type.Warning,
+                    FORCE_DELETED_EVENT_KEY,
+                    String.format(
+                            "Force deleted %d pod(s) that did not terminate in 
time "
+                                    + "(likely their node became unreachable), 
blocking cluster resource cleanup.",
+                            stuckPods.size()),
+                    EventRecorder.Component.JobManagerDeployment,

Review Comment:
   Would it be better to record this event at the CR level?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to