add ingress annotation: kubernetes.io/ingress.class

apache · Aug 31, 2023 · 2fa66e7 · 2fa66e7
1 parent 81c73c3
commit 2fa66e7
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 2 deletions.
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala
@@ -56,6 +56,18 @@ object K8sFlinkConfig {
     description = "retained tracking time for SILENT state flink tasks"
   )
 
+  /**
+   * If an ingress controller is specified in the configuration, the ingress class
+   * kubernetes.io/ingress.class must be specified when creating the ingress, since there are often
+   * multiple ingress controllers in a production environment.
+   */
+  val ingressClass: InternalOption = InternalOption(
+    key = "streampark.flink-k8s.ingress.class",
+    defaultValue = "streampark",
+    classType = classOf[java.lang.String],
+    description = "Direct ingress to the ingress controller."
+  )
+
   /** kubernetes default namespace */
   val DEFAULT_KUBERNETES_NAMESPACE = "default"
 

diff --git a/streampark-console/streampark-console-service/src/main/resources/application.yml b/streampark-console/streampark-console-service/src/main/resources/application.yml
@@ -121,6 +121,9 @@ streampark:
       polling-interval-sec:
         job-status: 2
         cluster-metric: 3
+    # If you need to specify an ingress controller, you can use this.
+    ingress:
+      class: nginx
 
   # packer garbage resources collection configuration
   packer-gc:

diff --git a/...netes/src/main/scala/org/apache/streampark/flink/kubernetes/ingress/IngressStrategy.scala b/...netes/src/main/scala/org/apache/streampark/flink/kubernetes/ingress/IngressStrategy.scala
@@ -17,7 +17,7 @@
 
 package org.apache.streampark.flink.kubernetes.ingress
 
-import org.apache.streampark.common.conf.ConfigConst
+import org.apache.streampark.common.conf.{ConfigConst, InternalConfigHolder, K8sFlinkConfig}
 
 import io.fabric8.kubernetes.api.model.{OwnerReference, OwnerReferenceBuilder}
 import io.fabric8.kubernetes.client.DefaultKubernetesClient
@@ -48,11 +48,16 @@ trait IngressStrategy {
   }
 
   def buildIngressAnnotations(clusterId: String): Map[String, String] = {
-    Map(
+    val annotations = Map(
       "nginx.ingress.kubernetes.io/rewrite-target" -> "/$2",
       "nginx.ingress.kubernetes.io/proxy-body-size" -> "1024m",
       "nginx.ingress.kubernetes.io/configuration-snippet" -> ("rewrite ^(/" + clusterId + ")$ $1/ permanent;")
     )
+    val ingressClass = InternalConfigHolder.get[String](K8sFlinkConfig.ingressClass)
+    if (ingressClass.nonEmpty) {
+      annotations + ("kubernetes.io/ingress.class" -> ingressClass)
+    }
+    annotations
   }
 
   def buildIngressLabels(clusterId: String): Map[String, String] = {