Skip to content
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

[Feature-4140][gateway] Flink Kubernetes opoerator supports ingress #4155

Merged
merged 1 commit into from
Jan 17, 2025
Merged

[Feature-4140][gateway] Flink Kubernetes opoerator supports ingress #4155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 4 additions & 0 deletions dinky-client/dinky-client-base/src/main/java/org/dinky/data/model/JarSubmitParam.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,4 +106,8 @@ public String getStatement() {
getArgs(),
getAllowNonRestoredState());
}

public String toString() {
return this.getStatement();
}
}
36 changes: 24 additions & 12 deletions ...main/java/org/dinky/gateway/kubernetes/operator/KubernetesApplicationOperatorGateway.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.dinky.utils.JsonUtils;
import org.dinky.utils.LogUtil;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;

import java.util.Collections;
Expand Down Expand Up @@ -127,19 +128,25 @@ public GatewayResult submitJar(FlinkUdfPathContextHolder udfPathContextHolder) {
// sleep a time ,because some time the service will not be found
Thread.sleep(3000);

// get jobmanager addr by service
ListOptions options = new ListOptions();
String serviceName = config.getFlinkConfig().getJobName() + "-rest";
options.setFieldSelector("metadata.name=" + serviceName);
ServiceList list = kubernetesClient
.services()
// fixed bug can't find service list #3700
.inNamespace(configuration.get(KubernetesConfigOptions.NAMESPACE))
.list(options);
if (Objects.nonNull(list) && list.getItems().isEmpty()) {
throw new RuntimeException("service list is empty, please check svc list is exists");
ServiceList list = null;
String ipPort = null;
if (!(StringUtils.isNotEmpty(getIngressUrl()) && pingIpPort(getIngressUrl()))) {
// get jobmanager addr by service
ListOptions options = new ListOptions();
String serviceName = config.getFlinkConfig().getJobName() + "-rest";
options.setFieldSelector("metadata.name=" + serviceName);
list = kubernetesClient
.services()
// fixed bug can't find service list #3700
.inNamespace(configuration.get(KubernetesConfigOptions.NAMESPACE))
.list(options);
if (Objects.nonNull(list) && list.getItems().isEmpty()) {
throw new RuntimeException("service list is empty, please check svc list is exists");
}
ipPort = getWebUrl(list, kubernetesClient);
} else {
ipPort = getIngressUrl();
}
String ipPort = getWebUrl(list, kubernetesClient);
result.setWebURL("http://" + ipPort);
result.setId(result.getJids().get(0) + System.currentTimeMillis());
result.success();
Expand All @@ -163,6 +170,11 @@ public String getWebUrl(ServiceList list, KubernetesClient kubernetesClient) {
StringBuilder ipPort = new StringBuilder();
StringBuilder svcRestPort = new StringBuilder();
StringBuilder svcType = new StringBuilder();

if (StringUtils.isNotEmpty(getIngressUrl()) && pingIpPort(getIngressUrl())) {
return getIngressUrl();
}

logger.debug("kubernetes service list : [{}] \n kubernetesClient: [{}]", list, kubernetesClient);
for (Service item : list.getItems()) {
svcRestPort
Expand Down
48 changes: 48 additions & 0 deletions ...ateway/src/main/java/org/dinky/gateway/kubernetes/operator/KubernetesOperatorGateway.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,17 +27,22 @@
import org.dinky.gateway.kubernetes.operator.api.AbstractPodSpec.Resource;
import org.dinky.gateway.kubernetes.operator.api.FlinkDeployment;
import org.dinky.gateway.kubernetes.operator.api.FlinkDeploymentSpec;
import org.dinky.gateway.kubernetes.operator.api.IngressSpec;
import org.dinky.gateway.kubernetes.operator.api.JobSpec;
import org.dinky.gateway.kubernetes.utils.DinkyKubernetsConstants;
import org.dinky.gateway.result.SavePointResult;
import org.dinky.gateway.result.TestResult;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand All @@ -48,6 +53,7 @@

import com.google.common.collect.Lists;

import cn.hutool.core.text.StrFormatter;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import lombok.Data;
import lombok.EqualsAndHashCode;
Expand Down Expand Up @@ -220,6 +226,8 @@ private void initSpec() {
flinkDeploymentSpec.setImage(image);
flinkDeployment.setSpec(flinkDeploymentSpec);

flinkDeploymentSpec.setIngress(this.buildIngress());

if (Asserts.isNotNull(serviceAccount)) {
flinkDeploymentSpec.setServiceAccount(serviceAccount);
logger.info("serviceAccount is : {}", serviceAccount);
Expand All @@ -229,6 +237,46 @@ private void initSpec() {
}
}

private IngressSpec buildIngress() {
String ingressDomain = checkUseIngress();
if (StringUtils.isNotEmpty(ingressDomain)) {
IngressSpec ingressSpec = new IngressSpec();
ingressSpec.setTemplate(ingressDomain + "/{{namespace}}/{{name}}(/|$)(.*)");
ingressSpec.setClassName("nginx");
HashMap annotations = new HashMap<String, String>();
annotations.put("nginx.ingress.kubernetes.io/rewrite-target", "/$2");
ingressSpec.setAnnotations(annotations);
return ingressSpec;
}
return null;
}

private String checkUseIngress() {
Map<String, Object> ingressConfig = k8sConfig.getIngressConfig();
if (MapUtils.isNotEmpty(ingressConfig)) {
boolean ingressEnable = Boolean.parseBoolean(ingressConfig
.getOrDefault(DinkyKubernetsConstants.DINKY_K8S_INGRESS_ENABLED_KEY, "false")
.toString());
String ingressDomain = ingressConfig
.getOrDefault(DinkyKubernetsConstants.DINKY_K8S_INGRESS_DOMAIN_KEY, StringUtils.EMPTY)
.toString();
if (ingressEnable && StringUtils.isNotEmpty(ingressDomain)) {
return ingressDomain;
}
}
return StringUtils.EMPTY;
}

protected String getIngressUrl() {
String jobName = config.getFlinkConfig().getJobName();
String nameSpace = kubernetesConfiguration.get("kubernetes.namespace");
String ingressDomain = checkUseIngress();
if (StringUtils.isNotEmpty(ingressDomain)) {
return StrFormatter.format("{}/{}/{}", ingressDomain, nameSpace, jobName);
}
return StringUtils.EMPTY;
}

private void initBase() {
flinkDeployment.setApiVersion("flink.apache.org/v1beta1");
flinkDeployment.setKind("FlinkDeployment");
Expand Down
4 changes: 2 additions & 2 deletions .../Cluster/Configuration/components/ConfigurationModal/ConfigurationForm/FlinkK8s/index.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ export default (props: {
placeholder={l('rc.cc.flinkConfigPathPlaceholder')}
tooltip={l('rc.cc.flinkConfigPathHelp')}
/>
{type && type === ClusterType.KUBERNETES_APPLICATION && (
{type && (type === ClusterType.KUBERNETES_APPLICATION || type === ClusterType.KUBERNETES_OPERATOR) && (
<ProFormSwitch
name={['config', 'kubernetesConfig', 'ingressConfig', 'kubernetes.ingress.enabled']}
label={l('rc.cc.k8s.ingress.enabled')}
Expand All @@ -210,7 +210,7 @@ export default (props: {
unCheckedChildren={l('button.disable')}
/>
)}
{type && type === ClusterType.KUBERNETES_APPLICATION && ingressEnabled && (
{type && (type === ClusterType.KUBERNETES_APPLICATION || type === ClusterType.KUBERNETES_OPERATOR) && ingressEnabled && (
<ProFormText
tooltip={l('rc.cc.k8s.ingress.domainHelp')}
name={['config', 'kubernetesConfig', 'ingressConfig', 'kubernetes.ingress.domain']}
Expand Down
Loading