You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/20 06:11:56 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27686] Only patch status on change
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 0e53730 [FLINK-27686] Only patch status on change
0e53730 is described below
commit 0e537308975f29a6dc129d047f3db337122cc3ca
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu May 19 15:31:35 2022 +0200
[FLINK-27686] Only patch status on change
---
.../kubernetes/operator/utils/StatusHelper.java | 21 ++++++--
.../kubernetes/operator/TestingStatusHelper.java | 6 ++-
.../operator/utils/StatusHelperTest.java | 59 ++++++++++++++++++++++
3 files changed, 80 insertions(+), 6 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusHelper.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusHelper.java
index aeacea1..ca96244 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusHelper.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusHelper.java
@@ -20,8 +20,9 @@ package org.apache.flink.kubernetes.operator.utils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.kubernetes.operator.crd.status.CommonStatus;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.client.CustomResource;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -36,7 +37,9 @@ public class StatusHelper<STATUS extends CommonStatus<?>> {
private static final Logger LOG = LoggerFactory.getLogger(StatusHelper.class);
- protected final ConcurrentHashMap<Tuple2<String, String>, STATUS> statusCache =
+ protected final ObjectMapper objectMapper = new ObjectMapper();
+
+ protected final ConcurrentHashMap<Tuple2<String, String>, ObjectNode> statusCache =
new ConcurrentHashMap<>();
private final KubernetesClient client;
@@ -64,6 +67,14 @@ public class StatusHelper<STATUS extends CommonStatus<?>> {
// in the meantime
resource.getMetadata().setResourceVersion(null);
+ ObjectNode newStatus = objectMapper.convertValue(resource.getStatus(), ObjectNode.class);
+ ObjectNode previousStatus = statusCache.put(getKey(resource), newStatus);
+
+ if (newStatus.equals(previousStatus)) {
+ LOG.debug("No status change.");
+ return;
+ }
+
Exception err = null;
for (int i = 0; i < 3; i++) {
// In any case we retry the status update 3 times to avoid some intermittent
@@ -73,7 +84,6 @@ public class StatusHelper<STATUS extends CommonStatus<?>> {
.inNamespace(namespace)
.withName(name)
.patchStatus(resource);
- statusCache.put(getKey(resource), ReconciliationUtils.clone(resource.getStatus()));
return;
} catch (Exception e) {
LOG.error("Error while patching status, retrying {}/3...", (i + 1), e);
@@ -97,7 +107,10 @@ public class StatusHelper<STATUS extends CommonStatus<?>> {
public <T extends CustomResource<?, STATUS>> void updateStatusFromCache(T resource) {
var cachedStatus = statusCache.get(getKey(resource));
if (cachedStatus != null) {
- resource.setStatus(ReconciliationUtils.clone(cachedStatus));
+ resource.setStatus(
+ (STATUS)
+ objectMapper.convertValue(
+ cachedStatus, resource.getStatus().getClass()));
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusHelper.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusHelper.java
index 361db94..ed29596 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusHelper.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusHelper.java
@@ -19,9 +19,9 @@
package org.apache.flink.kubernetes.operator;
import org.apache.flink.kubernetes.operator.crd.status.CommonStatus;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.fabric8.kubernetes.client.CustomResource;
/** Testing StatusHelper. */
@@ -33,6 +33,8 @@ public class TestingStatusHelper<STATUS extends CommonStatus<?>> extends StatusH
@Override
public <T extends CustomResource<?, STATUS>> void patchAndCacheStatus(T resource) {
- statusCache.put(getKey(resource), ReconciliationUtils.clone(resource.getStatus()));
+ statusCache.put(
+ getKey(resource),
+ objectMapper.convertValue(resource.getStatus(), ObjectNode.class));
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusHelperTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusHelperTest.java
new file mode 100644
index 0000000..8527f93
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusHelperTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link StatusHelper}. */
+@EnableKubernetesMockClient(crud = true)
+public class StatusHelperTest {
+
+ private KubernetesClient kubernetesClient;
+ private KubernetesMockServer mockServer;
+
+ @Test
+ public void testPatchOnlyWhenChanged() throws InterruptedException {
+ var helper = new StatusHelper<FlinkDeploymentStatus>(kubernetesClient);
+ var deployment = TestUtils.buildApplicationCluster();
+ kubernetesClient.resource(deployment).createOrReplace();
+ var lastRequest = mockServer.getLastRequest();
+
+ helper.patchAndCacheStatus(deployment);
+ assertTrue(mockServer.getLastRequest() != lastRequest);
+ lastRequest = mockServer.getLastRequest();
+ deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.UPGRADING);
+ helper.patchAndCacheStatus(deployment);
+
+ // We intentionally compare references
+ assertTrue(mockServer.getLastRequest() != lastRequest);
+ lastRequest = mockServer.getLastRequest();
+
+ // No update
+ helper.patchAndCacheStatus(deployment);
+ assertTrue(mockServer.getLastRequest() == lastRequest);
+ }
+}