[DSIP-76] Support ssl in netty

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertServer.java

@@ -25,6 +25,7 @@
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.registry.api.RegistryConfiguration;
 
 import javax.annotation.PostConstruct;
@@ -40,7 +41,8 @@
 @Slf4j
 @Import({CommonConfiguration.class,
         DaoConfiguration.class,
-        RegistryConfiguration.class})
+        RegistryConfiguration.class,
+        NettySslConfig.class})
 @SpringBootApplication
 public class AlertServer {
 

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/config/AlertConfig.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.alert.config;
 
 import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -47,6 +48,8 @@ public final class AlertConfig implements Validator {
 
     private String alertServerAddress;
 
+    private NettySslConfig nettySslConfig;
+
     @Override
     public boolean supports(Class<?> clazz) {
         return AlertConfig.class.isAssignableFrom(clazz);

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/resources/application.yaml

@@ -75,6 +75,11 @@ alert:
   max-heartbeat-interval: 60s
   # The maximum number of alerts that can be processed in parallel
   sender-parallelism: 100
+  rpc:
+    ssl:
+      enabled: false
+      cert-file-path: /path/cert.crt
+      key-file-path: /path/private.pem
 
 registry:
   type: zookeeper

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/ApiApplicationServer.java

@@ -23,6 +23,7 @@
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
 import org.apache.dolphinscheduler.dao.PluginDao;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
 import org.apache.dolphinscheduler.plugin.storage.api.StorageConfiguration;
 import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
@@ -44,14 +45,18 @@
         CommonConfiguration.class,
         ServiceConfiguration.class,
         StorageConfiguration.class,
-        RegistryConfiguration.class})
+        RegistryConfiguration.class,
+        NettySslConfig.class})
 @ServletComponentScan
 @SpringBootApplication
 public class ApiApplicationServer {
 
     @Autowired
     private PluginDao pluginDao;
 
+    @Autowired
+    NettySslConfig nettySslConfig;
+
     public static void main(String[] args) {
         ApiServerMetrics.registerUncachedException(DefaultUncaughtExceptionHandler::getUncaughtExceptionCount);
         Thread.setDefaultUncaughtExceptionHandler(DefaultUncaughtExceptionHandler.getInstance());

dolphinscheduler-api/src/main/resources/application.yaml

@@ -164,6 +164,11 @@ api:
     # Close each active connection of socket server if python program not active after x milliseconds. Define value is
     # (0 = infinite), and socket server would never close even though no requests accept
     read-timeout: 0
+  rpc:
+    ssl:
+      enabled: false
+      cert-file-path: /path/cert.crt
+      key-file-path: /path/private.pem
 
 metrics:
   enabled: true

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyRemotingClient.java

@@ -21,6 +21,7 @@
 import org.apache.dolphinscheduler.extract.base.IRpcResponse;
 import org.apache.dolphinscheduler.extract.base.SyncRequestDto;
 import org.apache.dolphinscheduler.extract.base.config.NettyClientConfig;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.extract.base.exception.RemotingException;
 import org.apache.dolphinscheduler.extract.base.exception.RemotingTimeoutException;
 import org.apache.dolphinscheduler.extract.base.future.ResponseFuture;
@@ -33,6 +34,7 @@
 import org.apache.dolphinscheduler.extract.base.utils.Host;
 import org.apache.dolphinscheduler.extract.base.utils.NettyUtils;
 
+import java.io.File;
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -41,6 +43,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 
+import javax.net.ssl.SSLException;
+
 import lombok.extern.slf4j.Slf4j;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
@@ -52,6 +56,8 @@
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.timeout.IdleStateHandler;
 
 @Slf4j
@@ -70,8 +76,20 @@ public class NettyRemotingClient implements AutoCloseable {
 
     private final NettyClientHandler clientHandler;
 
+    private SslContext sslContext = null;
+
     public NettyRemotingClient(final NettyClientConfig clientConfig) {
         this.clientConfig = clientConfig;
+        NettySslConfig nettySslConfig = clientConfig.getNettySslConfig();
+        if (nettySslConfig.isEnabled()) {
+            try {
+                sslContext =
+                        SslContextBuilder.forClient().trustManager(new File(nettySslConfig.getCertFilePath())).build();
+            } catch (SSLException e) {
+                throw new IllegalArgumentException("Initialize SslContext error, please check the cert-file", e);
+            }
+        }
+
         ThreadFactory nettyClientThreadFactory = ThreadUtils.newDaemonThreadFactory("NettyClientThread-");
         if (Epoll.isAvailable()) {
             this.workerGroup = new EpollEventLoopGroup(clientConfig.getWorkerThreads(), nettyClientThreadFactory);
@@ -97,6 +115,10 @@ private void start() {
 
                     @Override
                     public void initChannel(SocketChannel ch) {
+                        NettySslConfig nettySslConfig = clientConfig.getNettySslConfig();
+                        if (nettySslConfig.isEnabled()) {
+                            ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
+                        }
                         ch.pipeline()
                                 .addLast("client-idle-handler",
                                         new IdleStateHandler(

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyClientConfig.java

@@ -76,4 +76,7 @@ public class NettyClientConfig {
     @Builder.Default
     private int defaultRpcTimeoutMillis = 10_000;
 
+    @Builder.Default
+    private NettySslConfig nettySslConfig = new NettySslConfig();
+
 }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyServerConfig.java

@@ -79,4 +79,9 @@ public class NettyServerConfig {
      */
     private int listenPort;
 
+    /**
+     * nettySslConfig
+     */
+    @Builder.Default
+    private NettySslConfig nettySslConfig = new NettySslConfig();
 }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettySslConfig.java

@@ -0,0 +1,34 @@
+/*
+ * 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.dolphinscheduler.extract.base.config;
+
+import lombok.Data;
+
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@Data
+public class NettySslConfig {
+
+    public boolean enabled;
+
+    public String certFilePath;
+
+    public String keyFilePath;
+
+}

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/NettyRemotingServer.java

@@ -19,16 +19,20 @@
 
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.extract.base.config.NettyServerConfig;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.extract.base.exception.RemoteException;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterDecoder;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterEncoder;
 import org.apache.dolphinscheduler.extract.base.utils.NettyUtils;
 
+import java.io.File;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.net.ssl.SSLException;
+
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import io.netty.bootstrap.ServerBootstrap;
@@ -40,8 +44,9 @@
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.timeout.IdleStateHandler;
-
 /**
  * remoting netty server
  */
@@ -66,7 +71,18 @@ class NettyRemotingServer {
 
     private final AtomicBoolean isStarted = new AtomicBoolean(false);
 
-    NettyRemotingServer(final NettyServerConfig serverConfig) {
+    private SslContext sslContext = null;
+
+    public NettyRemotingServer(final NettyServerConfig serverConfig) {
+        NettySslConfig nettySslConfig = serverConfig.getNettySslConfig();
+        if (nettySslConfig.isEnabled()) {
+            try {
+                sslContext = SslContextBuilder.forServer(new File(nettySslConfig.getCertFilePath()),
+                        new File(nettySslConfig.getKeyFilePath())).build();
+            } catch (SSLException e) {
+                throw new RuntimeException(e);
+            }
+        }
         this.serverConfig = serverConfig;
         this.serverName = serverConfig.getServerName();
         this.methodInvokerExecutor = ThreadUtils.newDaemonFixedThreadExecutor(
@@ -130,6 +146,10 @@ protected void initChannel(SocketChannel ch) {
      * @param ch socket channel
      */
     private void initNettyChannel(SocketChannel ch) {
+        NettySslConfig nettySslConfig = serverConfig.getNettySslConfig();
+        if (nettySslConfig.isEnabled()) {
+            ch.pipeline().addLast("ssl", sslContext.newHandler(ch.alloc()));
+        }
         ch.pipeline()
                 .addLast("encoder", new TransporterEncoder())
                 .addLast("decoder", new TransporterDecoder())

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java

@@ -24,6 +24,7 @@
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
 import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
@@ -60,7 +61,8 @@
         ServiceConfiguration.class,
         CommonConfiguration.class,
         StorageConfiguration.class,
-        RegistryConfiguration.class})
+        RegistryConfiguration.class,
+        NettySslConfig.class})
 @SpringBootApplication
 public class MasterServer implements IStoppable {
 
@@ -94,6 +96,9 @@ public class MasterServer implements IStoppable {
     @Autowired
     private SystemEventBusFireWorker systemEventBusFireWorker;
 
+    @Autowired
+    NettySslConfig nettySslConfig;
+
     public static void main(String[] args) {
         MasterServerMetrics.registerUncachedException(DefaultUncaughtExceptionHandler::getUncaughtExceptionCount);
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.config;
 
 import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.registry.api.ConnectStrategyProperties;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.WorkerLoadBalancerConfigurationProperties;
@@ -79,6 +80,8 @@ public class MasterConfig implements Validator {
      */
     private String masterRegistryPath;
 
+    private NettySslConfig nettySslConfig;
+
     @Override
     public boolean supports(Class<?> clazz) {
         return MasterConfig.class.isAssignableFrom(clazz);

dolphinscheduler-master/src/main/resources/application.yaml

@@ -121,6 +121,11 @@ master:
       memory-usage-weight: 30
       cpu-usage-weight: 30
       task-thread-pool-usage-weight: 40
+  rpc:
+    ssl:
+      enabled: false
+      cert-file-path: /path/cert.crt
+      key-file-path: /path/private.pem
 
 server:
   port: 5679

dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/WorkerServer.java

@@ -23,6 +23,7 @@
 import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
 import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
@@ -55,7 +56,8 @@
 @Slf4j
 @Import({CommonConfiguration.class,
         StorageConfiguration.class,
-        RegistryConfiguration.class})
+        RegistryConfiguration.class,
+        NettySslConfig.class})
 @SpringBootApplication
 public class WorkerServer implements IStoppable {
 
@@ -71,6 +73,8 @@ public class WorkerServer implements IStoppable {
     @Autowired
     private MetricsProvider metricsProvider;
 
+    @Autowired
+    NettySslConfig nettySslConfig;
     /**
      * worker server startup, not use web service
      *

dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.worker.config;
 
 import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.extract.base.config.NettySslConfig;
 import org.apache.dolphinscheduler.registry.api.ConnectStrategyProperties;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 
@@ -47,7 +48,7 @@ public class WorkerConfig implements Validator {
     private int hostWeight = 100;
     private WorkerServerLoadProtection serverLoadProtection = new WorkerServerLoadProtection();
     private ConnectStrategyProperties registryDisconnectStrategy = new ConnectStrategyProperties();
-
+    private NettySslConfig nettySslConfig;
     /**
      * This field doesn't need to set at config file, it will be calculated by workerIp:listenPort
      */

dolphinscheduler-worker/src/main/resources/application.yaml

@@ -68,6 +68,11 @@ worker:
     auto-create-tenant-enabled: true
     # If set true, will use worker bootstrap user as the tenant to execute task when the tenant is `default`.
     default-tenant-enabled: false
+  rpc:
+    ssl:
+      enabled: false
+      cert-file-path: /path/cert.crt
+      key-file-path: /path/private.pem
 
 server:
   port: 1235