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/17 07:05:34 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27612] Generate warning events when deleting the session cluster with running jobs
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 7ac7e99 [FLINK-27612] Generate warning events when deleting the session cluster with running jobs
7ac7e99 is described below
commit 7ac7e991c51ecd906c86544deca038a128e413b9
Author: Aitozi <yu...@alibaba-inc.com>
AuthorDate: Tue May 17 15:05:30 2022 +0800
[FLINK-27612] Generate warning events when deleting the session cluster with running jobs
---
.../reconciler/deployment/SessionReconciler.java | 12 ++-
.../kubernetes/operator/utils/EventUtils.java | 112 +++++++++++++++++++++
.../kubernetes/operator/utils/EventUtilsTest.java | 84 ++++++++++++++++
3 files changed, 206 insertions(+), 2 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index 9d3ea90..246b276 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -28,6 +28,7 @@ import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import org.apache.flink.kubernetes.operator.utils.EventUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.IngressUtils;
import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
@@ -161,8 +162,15 @@ public class SessionReconciler extends AbstractDeploymentReconciler {
sessionJobs.stream()
.map(job -> job.getMetadata().getName())
.collect(Collectors.toList()));
- // TODO generate error events for this
- LOG.warn(error);
+ if (EventUtils.createOrUpdateEvent(
+ kubernetesClient,
+ flinkApp,
+ EventUtils.Type.Warning,
+ "Cleanup",
+ error,
+ EventUtils.Component.Operator)) {
+ LOG.warn(error);
+ }
return DeleteControl.noFinalizerRemoval()
.rescheduleAfter(
configManager
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java
new file mode 100644
index 0000000..c1ec722
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java
@@ -0,0 +1,112 @@
+/*
+ * 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 io.fabric8.kubernetes.api.model.EventBuilder;
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectReferenceBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+
+import java.time.Instant;
+
+/**
+ * The util to generate an event for the target resource. It is copied from
+ * https://github.com/EnMasseProject/enmasse/blob/master/k8s-api/src/main/java/io/enmasse/k8s/api/KubeEventLogger.java
+ */
+public class EventUtils {
+
+ /** The type of the events. */
+ public enum Type {
+ Normal,
+ Warning
+ }
+
+ /** The component of events. */
+ public enum Component {
+ Operator
+ }
+
+ public static String generateEventName(
+ HasMetadata target, Type type, String reason, String message, Component component) {
+ return component
+ + "."
+ + ((reason + message + type + target.getKind() + target.getMetadata().getName())
+ .hashCode()
+ & 0x7FFFFFFF);
+ }
+
+ public static boolean createOrUpdateEvent(
+ KubernetesClient client,
+ HasMetadata target,
+ Type type,
+ String reason,
+ String message,
+ Component component) {
+ var eventName = generateEventName(target, type, reason, message, component);
+
+ var existing =
+ client.v1()
+ .events()
+ .inNamespace(target.getMetadata().getNamespace())
+ .withName(eventName)
+ .get();
+
+ if (existing != null
+ && existing.getType().equals(type.name())
+ && existing.getReason().equals(reason)
+ && existing.getInvolvedObject().getName().equals(target.getMetadata().getName())
+ && existing.getInvolvedObject().getKind().equals(target.getKind())) {
+ // update
+ existing.setLastTimestamp(Instant.now().toString());
+ existing.setCount(existing.getCount() + 1);
+ existing.setMessage(message);
+ client.v1()
+ .events()
+ .inNamespace(target.getMetadata().getNamespace())
+ .createOrReplace(existing);
+ return false;
+ } else {
+ var event =
+ new EventBuilder()
+ .withApiVersion("v1")
+ .withInvolvedObject(
+ new ObjectReferenceBuilder()
+ .withKind(target.getKind())
+ .withUid(target.getMetadata().getUid())
+ .withName(target.getMetadata().getName())
+ .withNamespace(target.getMetadata().getNamespace())
+ .build())
+ .withType(type.name())
+ .withReason(reason)
+ .withFirstTimestamp(Instant.now().toString())
+ .withLastTimestamp(Instant.now().toString())
+ .withNewSource()
+ .withComponent(component.name())
+ .endSource()
+ .withCount(1)
+ .withMessage(message)
+ .withNewMetadata()
+ .withName(eventName)
+ .withNamespace(target.getMetadata().getNamespace())
+ .endMetadata()
+ .build();
+ client.v1().events().inNamespace(target.getMetadata().getNamespace()).create(event);
+ return true;
+ }
+ }
+}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java
new file mode 100644
index 0000000..bc3a07f
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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 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.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Test for {@link EventUtils}. */
+@EnableKubernetesMockClient(crud = true)
+public class EventUtilsTest {
+
+ private KubernetesMockServer mockServer;
+ private KubernetesClient kubernetesClient;
+
+ @Test
+ public void testCreateOrReplaceEvent() {
+ var flinkApp = TestUtils.buildApplicationCluster();
+ var reason = "Cleanup";
+ var message = "message";
+ var eventName =
+ EventUtils.generateEventName(
+ flinkApp,
+ EventUtils.Type.Warning,
+ reason,
+ message,
+ EventUtils.Component.Operator);
+ Assertions.assertTrue(
+ EventUtils.createOrUpdateEvent(
+ kubernetesClient,
+ flinkApp,
+ EventUtils.Type.Warning,
+ reason,
+ message,
+ EventUtils.Component.Operator));
+ var event =
+ kubernetesClient
+ .v1()
+ .events()
+ .inNamespace(flinkApp.getMetadata().getNamespace())
+ .withName(eventName)
+ .get();
+ Assertions.assertNotNull(event);
+ Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(reason, event.getReason());
+
+ Assertions.assertFalse(
+ EventUtils.createOrUpdateEvent(
+ kubernetesClient,
+ flinkApp,
+ EventUtils.Type.Warning,
+ reason,
+ message,
+ EventUtils.Component.Operator));
+ event =
+ kubernetesClient
+ .v1()
+ .events()
+ .inNamespace(flinkApp.getMetadata().getNamespace())
+ .withName(eventName)
+ .get();
+
+ Assertions.assertEquals(2, event.getCount());
+ }
+}