Add Flink log collector and the transform in streamis which resolves …

…the internal configuration

pom.xml

@@ -46,6 +46,7 @@
 
     <properties>
         <linkis.version>1.1.3</linkis.version>
+        <junit.version>4.12</junit.version>
         <dss.version>1.1.0</dss.version>
         <streamis.version>0.2.0</streamis.version>
         <scala.version>2.11.12</scala.version>

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/pom.xml

@@ -16,13 +16,80 @@
         <maven.compiler.target>8</maven.compiler.target>
         <!-- The flink version dependent-->
         <flink.version>1.12.2</flink.version>
+        <log4j.version>2.17.1</log4j.version>
+        <slf4j.version>1.7.15</slf4j.version>
     </properties>
     <dependencies>
+        <dependency>
+            <groupId>com.webank.wedatasphere.streamis</groupId>
+            <artifactId>streamis-job-log-collector</artifactId>
+            <version>${streamis.version}</version>
+        </dependency>
         <!-- flink basic dependencies-->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-java</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-yarn_2.11</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <!--Junit-->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <!--log4j2-->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <version>${log4j.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>
+            <version>${log4j.version}</version>
+            <scope>provided</scope>
         </dependency>
     </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <version>2.3</version>
+                <executions>
+                    <execution>
+                        <id>assemble</id>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                        <!-- install -->
+                        <phase>install</phase>
+                    </execution>
+                </executions>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <appendAssemblyId>false</appendAssemblyId>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/assembly/package.xml

@@ -0,0 +1,19 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
+
+    <id>package</id>
+
+    <formats>
+        <format>jar</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <dependencySets>
+        <dependencySet>
+            <outputDirectory>/</outputDirectory>
+            <unpack>true</unpack>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkStreamisConfigAutowired.java

@@ -0,0 +1,104 @@
+package com.webank.wedatasphere.streamis.jobmanager.log.collector.flink;
+
+import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
+import com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.Filter;
+import org.apache.logging.log4j.core.filter.LevelMatchFilter;
+import org.apache.logging.log4j.core.filter.RegexFilter;
+
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Properties;
+
+import static com.webank.wedatasphere.streamis.jobmanager.log.collector.flink.FlinkStreamisConfigDefine.*;
+
+/**
+ * Autoconfigure the streamis config inf Flink environment
+ */
+public class FlinkStreamisConfigAutowired implements StreamisConfigAutowired {
+
+    /**
+     * Flink configuration
+     */
+    private Configuration configuration;
+
+    public FlinkStreamisConfigAutowired(){
+        // First to load configuration
+        this.configuration = loadConfiguration();
+    }
+    @Override
+    public StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Builder builder) throws Exception{
+        String applicationName = this.configuration.getString(YarnConfigOptions.APPLICATION_NAME);
+        if (StringUtils.isNotBlank(applicationName)){
+            builder.setAppName(applicationName);
+        }
+        String gateway = this.configuration.getString(LOG_GATEWAY_ADDRESS);
+        if (StringUtils.isNotBlank(gateway)){
+            if (gateway.endsWith("/")){
+                gateway = gateway.substring(0, gateway.length() - 1);
+            }
+            gateway += this.configuration.getString(LOG_COLLECT_PATH, "/");
+            builder.setRpcAddress(gateway);
+        }
+        List<String> filterStrategies = this.configuration.get(LOG_FILTER_STRATEGIES);
+        for(String filterStrategy : filterStrategies){
+            if ("LevelMatch".equals(filterStrategy)){
+                builder.withFilter(LevelMatchFilter.newBuilder()
+                        .setLevel(Level.getLevel(this.configuration.getString(LOG_FILTER_LEVEL_MATCH))).build());
+            } else if ("RegexMatch".equals(filterStrategy)){
+                builder.withFilter(RegexFilter.createFilter( this.configuration.getString(LOG_FILTER_REGEX),
+                        null, true, Filter.Result.ACCEPT, Filter.Result.DENY));
+            }
+        }
+        return builder.setRpcConnTimeout(this.configuration.getInteger(LOG_RPC_CONN_TIMEOUT))
+                .setRpcSocketTimeout(this.configuration.getInteger(LOG_RPC_SOCKET_TIMEOUT))
+                .setRpcSendRetryCnt(this.configuration.getInteger(LOG_RPC_SEND_RETRY_COUNT))
+                .setRpcServerRecoveryTimeInSec(this.configuration.getInteger(LOG_RPC_SERVER_RECOVERY_TIME))
+                .setRpcMaxDelayTimeInSec(this.configuration.getInteger(LOG_RPC_MAX_DELAY_TIME))
+                .setRpcAuthTokenCodeKey(this.configuration.getString(LOG_RPC_AUTH_TOKEN_CODE_KEY))
+                .setRpcAuthTokenUserKey(this.configuration.getString(LOG_RPC_AUTH_TOKEN_USER_KEY))
+                .setRpcAuthTokenCode(this.configuration.getString(LOG_RPC_AUTH_TOKEN_CODE))
+                .setRpcAuthTokenUser(this.configuration.getString(LOG_RPC_AUTH_TOKEN_USER))
+                .setRpcCacheSize(this.configuration.getInteger(LOG_RPC_CACHE_SIZE))
+                .setRpcCacheMaxConsumeThread(this.configuration.getInteger(LOG_PRC_CACHE_MAX_CONSUME_THREAD))
+                .setRpcBufferSize(this.configuration.getInteger(LOG_RPC_BUFFER_SIZE))
+                .setRpcBufferExpireTimeInSec(this.configuration.getInteger(LOG_RPC_BUFFER_EXPIRE_TIME)).build();
+    }
+
+    /**
+     * According to :
+     * String launchCommand =
+     *                 BootstrapTools.getTaskManagerShellCommand(
+     *                         flinkConfig,
+     *                         tmParams,
+     *                         ".",
+     *                         ApplicationConstants.LOG_DIR_EXPANSION_VAR,
+     *                         hasLogback,
+     *                         hasLog4j,
+     *                         hasKrb5,
+     *                         taskManagerMainClass,
+     *                         taskManagerDynamicProperties);
+     * the configuration directory of Flink yarn container is always ".",
+     * @return configuration
+     */
+    private Configuration loadConfiguration(){
+        String configDir = System.getenv("FLINK_CONF_DIR");
+        if (null == configDir){
+            configDir = ".";
+        }
+        Properties properties = System.getProperties();
+        Enumeration<?> enumeration = properties.propertyNames();
+        Configuration dynamicConfiguration = new Configuration();
+        while(enumeration.hasMoreElements()){
+            String prop = String.valueOf(enumeration.nextElement());
+            dynamicConfiguration.setString(prop, properties.getProperty(prop));
+        }
+        return GlobalConfiguration.loadConfiguration(configDir, dynamicConfiguration);
+    }
+
+}

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkStreamisConfigDefine.java

@@ -0,0 +1,120 @@
+package com.webank.wedatasphere.streamis.jobmanager.log.collector.flink;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.util.List;
+
+/**
+ * Config definition
+ */
+public class FlinkStreamisConfigDefine {
+
+    /**
+     * Gateway address of log module for streamis
+     */
+    public static final ConfigOption<String> LOG_GATEWAY_ADDRESS = ConfigOptions.key("stream.log.gateway.address")
+            .stringType().noDefaultValue().withDescription("The gateway address ex: http://127.0.0.1:8080");
+
+    /**
+     * Entrypoint path of collecting log
+     */
+    public static final ConfigOption<String> LOG_COLLECT_PATH = ConfigOptions.key("stream.log.collect.path")
+            .stringType().defaultValue("/api/rest_j/v1/streamis/streamJobManager/log/collect/events").withDescription("The entrypoint path of collecting log");
+
+    /**
+     * Connection timeout(in milliseconds) in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_CONN_TIMEOUT = ConfigOptions.key("stream.log.rpc.connect-timeout")
+            .intType().defaultValue(3000).withDescription("Connection timeout(ms) in log RPC module");
+
+    /**
+     * Socket timeout(in milliseconds) in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_SOCKET_TIMEOUT = ConfigOptions.key("stream.log.rpc.socket-timeout")
+            .intType().defaultValue(15000).withDescription("Socket timeout(ms) in log RPC module");
+
+    /**
+     * Max retry count of sending message in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_SEND_RETRY_COUNT = ConfigOptions.key("stream.log.rpc.send-retry-count")
+            .intType().defaultValue(3).withDescription("Max retry count of sending message in log RPC module");
+
+    /**
+     * Server recovery time(in seconds) in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_SERVER_RECOVERY_TIME = ConfigOptions.key("stream.log.rpc.server-recovery-time-in-sec")
+            .intType().defaultValue(5).withDescription("Server recovery time(sec) in log RPC module");
+
+    /**
+     * Max delay time(in seconds) in log RPC module. if reach the limit, the message will be dropped
+     */
+    public static final ConfigOption<Integer> LOG_RPC_MAX_DELAY_TIME = ConfigOptions.key("stream.log.rpc.max-delay-time")
+            .intType().defaultValue(60).withDescription("Max delay time(sec) in log RPC module");
+
+    /**
+     * Token code key in log RPC auth module
+     */
+    public static final ConfigOption<String> LOG_RPC_AUTH_TOKEN_CODE_KEY = ConfigOptions.key("stream.log.rpc.auth.token-code-key")
+            .stringType().defaultValue("Token-Code").withDescription("Token code key in log RPC auth module");
+
+    /**
+     * Token user key in log RPC auth module
+     */
+    public static final ConfigOption<String> LOG_RPC_AUTH_TOKEN_USER_KEY = ConfigOptions.key("stream.log.rpc.auth.token-user-key")
+            .stringType().defaultValue("Token-User").withDescription("Token user key in log RPC auth module");
+
+    /**
+     * Token code in log RPC auth module
+     */
+    public static final ConfigOption<String> LOG_RPC_AUTH_TOKEN_CODE = ConfigOptions.key("stream.log.rpc.auth.token-code")
+            .stringType().defaultValue("STREAM-LOG").withDescription("Token code in log RPC auth module");
+
+    /**
+     * Token user in log RPC auth module
+     */
+    public static final ConfigOption<String> LOG_RPC_AUTH_TOKEN_USER = ConfigOptions.key("stream.log.rpc.auth.token-user")
+            .stringType().defaultValue(System.getProperty("user.name")).withDescription("Token user in log RPC auth module");
+
+    /**
+     * Cache size in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_CACHE_SIZE = ConfigOptions.key("stream.log.rpc.cache.size")
+            .intType().defaultValue(150).withDescription("Cache size in log RPC module");
+
+    /**
+     * Max cache consume threads in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_PRC_CACHE_MAX_CONSUME_THREAD = ConfigOptions.key("stream.log.rpc.cache.max-consume-thread")
+            .intType().defaultValue(10).withDescription("Max cache consume threads in log RPC module");
+
+    /**
+     * Buffer size in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_BUFFER_SIZE = ConfigOptions.key("stream.log.rpc.buffer.size")
+            .intType().defaultValue(50).withDescription("Buffer size in log RPC module");
+
+    /**
+     * Buffer expire time(sec) in log RPC module
+     */
+    public static final ConfigOption<Integer> LOG_RPC_BUFFER_EXPIRE_TIME = ConfigOptions.key("stream.log.rpc.buffer.expire-time-in-sec")
+            .intType().defaultValue(2).withDescription("Buffer expire time (sec) in log RPC module");
+
+    /**
+     * Log filter strategy list
+     */
+    public static final ConfigOption<List<String>> LOG_FILTER_STRATEGIES = ConfigOptions.key("stream.log.filter.strategies")
+            .stringType().asList().defaultValues("LevelMatch").withDescription("Log filter strategy list");
+
+    /**
+     * Level value of LevelMatch filter strategy
+     */
+    public static final ConfigOption<String> LOG_FILTER_LEVEL_MATCH = ConfigOptions.key("stream.log.filter.level-match.level")
+            .stringType().defaultValue("ERROR").withDescription("Level value of LevelMatch filter strategy");
+
+    /**
+     * Regex value of RegexMatch filter strategy
+     */
+    public static final ConfigOption<String> LOG_FILTER_REGEX = ConfigOptions.key("stream.log.filter.regex.value")
+            .stringType().defaultValue(".*").withDescription("Regex value of RegexMatch filter strategy");
+}

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/resources/META-INF/services/com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired

@@ -0,0 +1 @@
+com.webank.wedatasphere.streamis.jobmanager.log.collector.flink.FlinkStreamisConfigAutowired

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkConfigurationLoadTest.java

@@ -0,0 +1,25 @@
+package com.webank.wedatasphere.streamis.jobmanager.log.collector.flink;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.Enumeration;
+import java.util.Objects;
+import java.util.Properties;
+
+public class FlinkConfigurationLoadTest {
+    @Test
+    public void loadConfiguration() {
+        String configDir = Objects.requireNonNull(FlinkConfigurationLoadTest.class.getResource("/")).getFile();
+        Properties properties = System.getProperties();
+        Enumeration<?> enumeration = properties.propertyNames();
+        Configuration dynamicConfiguration = new Configuration();
+        while(enumeration.hasMoreElements()){
+            String prop = String.valueOf(enumeration.nextElement());
+            dynamicConfiguration.setString(prop, properties.getProperty(prop));
+        }
+        GlobalConfiguration.loadConfiguration(configDir, dynamicConfiguration);
+    }
+}

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/pom.xml

@@ -24,7 +24,7 @@
         <dependency>
             <groupId>com.webank.wedatasphere.streamis</groupId>
             <artifactId>streamis-job-log-common</artifactId>
-            <version>0.2.0</version>
+            <version>${streamis.version}</version>
         </dependency>
         <!--http client module-->
         <dependency>
@@ -65,5 +65,12 @@
             <version>${log4j.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisRpcLogAppender.java

@@ -4,7 +4,9 @@
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.RpcLogSenderConfig;
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.sender.StreamisRpcLogSender;
+import com.webank.wedatasphere.streamis.jobmanager.log.collector.sender.http.Json;
 import com.webank.wedatasphere.streamis.jobmanager.log.entities.StreamisLogEvent;
+import com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired;
 import org.apache.logging.log4j.core.Filter;
 import org.apache.logging.log4j.core.Layout;
 import org.apache.logging.log4j.core.LogEvent;
@@ -27,9 +29,7 @@
  */
 @Plugin(name = "StreamRpcLog", category = "Core", elementType = "appender", printObject = true)
 public class StreamisRpcLogAppender extends AbstractAppender {
-
     private static final String DEFAULT_APPENDER_NAME = "StreamRpcLog";
-
     /**
      * Appender config
      */
@@ -52,6 +52,8 @@ protected StreamisRpcLogAppender(String name, Filter filter,
         this.appenderConfig = appenderConfig;
         this.rpcLogSender = new StreamisRpcLogSender(this.appenderConfig.getApplicationName(),
                 this.appenderConfig.getSenderConfig());
+        this.rpcLogSender.setExceptionListener((subject, t, message) ->
+                LOGGER.error((null != subject? subject.getClass().getSimpleName() : "") + ": " + message, t));
         this.logCache = this.rpcLogSender.getOrCreateLogCache();
         Runtime.getRuntime().addShutdownHook(new Thread(this.rpcLogSender::close));
     }
@@ -74,7 +76,7 @@ public static StreamisRpcLogAppender createAppender(@PluginAttribute("name") Str
                                                         @PluginAttribute("ignoreExceptions") boolean ignoreExceptions,
                                                         @PluginElement("Filter") final Filter filter,
                                                         @PluginElement("Layout") Layout<? extends Serializable> layout,
-                                                        @PluginElement("RpcLogSender")RpcLogSenderConfig rpcLogSenderConfig){
+                                                         @PluginElement("RpcLogSender")RpcLogSenderConfig rpcLogSenderConfig) throws Exception{
         if (null == name || name.trim().equals("")){
             name = DEFAULT_APPENDER_NAME;
         }
@@ -97,6 +99,7 @@ public static StreamisRpcLogAppender createAppender(@PluginAttribute("name") Str
         if (null == applicationName || applicationName.trim().equals("")){
             throw new IllegalArgumentException("Application name cannot be empty");
         }
+        LOGGER.info("StreamisRpcLogAppender: init with config {}", Json.toJson(logAppenderConfig, null));
         return new StreamisRpcLogAppender(name, filter, layout, ignoreExceptions, Property.EMPTY_ARRAY, logAppenderConfig);
     }
 

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/StreamisLogAppenderConfig.java

@@ -56,7 +56,7 @@ public Builder(String applicationName, Filter filter,
          * @param applicationName application name
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setAppName(String applicationName){
+        public StreamisLogAppenderConfig.Builder setAppName(String applicationName){
             this.applicationName = applicationName;
             return this;
         }
@@ -66,7 +66,7 @@ StreamisLogAppenderConfig.Builder setAppName(String applicationName){
          * @param filter filter
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setFilter(Filter filter){
+        public StreamisLogAppenderConfig.Builder setFilter(Filter filter){
             this.filters.clear();
             this.filters.add(filter);
             return this;
@@ -77,17 +77,27 @@ StreamisLogAppenderConfig.Builder setFilter(Filter filter){
          * @param filter filter
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder withFilter(Filter filter){
+        public StreamisLogAppenderConfig.Builder withFilter(Filter filter){
             filters.add(filter);
             return this;
         }
 
+        /**
+         * Rpc address
+         * @param address address
+         * @return builder
+         */
+        public StreamisLogAppenderConfig.Builder setRpcAddress(String address){
+            this.rpcLogSenderConfig.setAddress(address);
+            return this;
+        }
+
         /**
          * Rpc connect timeout
          * @param connectionTimeout connection timeout
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcConnTimeout(int connectionTimeout){
+        public StreamisLogAppenderConfig.Builder setRpcConnTimeout(int connectionTimeout){
             this.rpcLogSenderConfig.setConnectionTimeout(connectionTimeout);
             return this;
         }
@@ -97,7 +107,7 @@ StreamisLogAppenderConfig.Builder setRpcConnTimeout(int connectionTimeout){
          * @param socketTimeout socket timeout
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcSocketTimeout(int socketTimeout){
+        public StreamisLogAppenderConfig.Builder setRpcSocketTimeout(int socketTimeout){
             this.rpcLogSenderConfig.setSocketTimeout(socketTimeout);
             return this;
         }
@@ -107,7 +117,7 @@ StreamisLogAppenderConfig.Builder setRpcSocketTimeout(int socketTimeout){
          * @param sendRetryCnt send retry count
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcSendRetryCnt(int sendRetryCnt){
+        public StreamisLogAppenderConfig.Builder setRpcSendRetryCnt(int sendRetryCnt){
             this.rpcLogSenderConfig.setSendRetryCnt(sendRetryCnt);
             return this;
         }
@@ -117,7 +127,7 @@ StreamisLogAppenderConfig.Builder setRpcSendRetryCnt(int sendRetryCnt){
          * @param serverRecoveryTimeInSec server recovery time
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcServerRecoveryTimeInSec(int serverRecoveryTimeInSec){
+        public StreamisLogAppenderConfig.Builder setRpcServerRecoveryTimeInSec(int serverRecoveryTimeInSec){
             this.rpcLogSenderConfig.setServerRecoveryTimeInSec(serverRecoveryTimeInSec);
             return this;
         }
@@ -127,7 +137,7 @@ StreamisLogAppenderConfig.Builder setRpcServerRecoveryTimeInSec(int serverRecove
          * @param maxDelayTimeInSec max delay time in seconds
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcMaxDelayTimeInSec(int maxDelayTimeInSec){
+        public StreamisLogAppenderConfig.Builder setRpcMaxDelayTimeInSec(int maxDelayTimeInSec){
             this.rpcLogSenderConfig.setMaxDelayTimeInSec(maxDelayTimeInSec);
             return this;
         }
@@ -137,7 +147,7 @@ StreamisLogAppenderConfig.Builder setRpcMaxDelayTimeInSec(int maxDelayTimeInSec)
          * @param tokenCodeKey key of token code
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcAuthTokenCodeKey(String tokenCodeKey){
+        public StreamisLogAppenderConfig.Builder setRpcAuthTokenCodeKey(String tokenCodeKey){
             this.rpcLogSenderConfig.getAuthConfig().setTokenCodeKey(tokenCodeKey);
             return this;
         }
@@ -147,7 +157,7 @@ StreamisLogAppenderConfig.Builder setRpcAuthTokenCodeKey(String tokenCodeKey){
          * @param tokenUserKey key of token user
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcAuthTokenUserKey(String tokenUserKey){
+        public StreamisLogAppenderConfig.Builder setRpcAuthTokenUserKey(String tokenUserKey){
             this.rpcLogSenderConfig.getAuthConfig().setTokenUserKey(tokenUserKey);
             return this;
         }
@@ -157,7 +167,7 @@ StreamisLogAppenderConfig.Builder setRpcAuthTokenUserKey(String tokenUserKey){
          * @param tokenUser token user
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcAuthTokenUser(String tokenUser){
+        public StreamisLogAppenderConfig.Builder setRpcAuthTokenUser(String tokenUser){
             this.rpcLogSenderConfig.getAuthConfig().setTokenUser(tokenUser);
             return this;
         }
@@ -167,7 +177,7 @@ StreamisLogAppenderConfig.Builder setRpcAuthTokenUser(String tokenUser){
          * @param tokenCode token code
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcAuthTokenCode(String tokenCode){
+        public StreamisLogAppenderConfig.Builder setRpcAuthTokenCode(String tokenCode){
             this.rpcLogSenderConfig.getAuthConfig().setTokenCode(tokenCode);
             return this;
         }
@@ -177,7 +187,7 @@ StreamisLogAppenderConfig.Builder setRpcAuthTokenCode(String tokenCode){
          * @param cacheSize cache size
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcCacheSize(int cacheSize){
+        public StreamisLogAppenderConfig.Builder setRpcCacheSize(int cacheSize){
             this.rpcLogSenderConfig.getCacheConfig().setSize(cacheSize);
             return this;
         }
@@ -187,7 +197,7 @@ StreamisLogAppenderConfig.Builder setRpcCacheSize(int cacheSize){
          * @param maxConsumeThread max consume thread
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcCacheMaxConsumeThread(int maxConsumeThread){
+        public StreamisLogAppenderConfig.Builder setRpcCacheMaxConsumeThread(int maxConsumeThread){
             this.rpcLogSenderConfig.getCacheConfig().setMaxConsumeThread(maxConsumeThread);
             return this;
         }
@@ -197,7 +207,7 @@ StreamisLogAppenderConfig.Builder setRpcCacheMaxConsumeThread(int maxConsumeThre
          * @param bufferSize buffer size
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcBufferSize(int bufferSize){
+        public StreamisLogAppenderConfig.Builder setRpcBufferSize(int bufferSize){
             this.rpcLogSenderConfig.getBufferConfig().setSize(bufferSize);
             return this;
         }
@@ -207,7 +217,7 @@ StreamisLogAppenderConfig.Builder setRpcBufferSize(int bufferSize){
          * @param expireTimeInSec expire time
          * @return builder
          */
-        StreamisLogAppenderConfig.Builder setRpcBufferExpireTimeInSec(int expireTimeInSec){
+        public StreamisLogAppenderConfig.Builder setRpcBufferExpireTimeInSec(int expireTimeInSec){
             this.rpcLogSenderConfig.getBufferConfig().setExpireTimeInSec(expireTimeInSec);
             return this;
         }

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/SendLogCacheConsumer.java

@@ -87,6 +87,7 @@ public void run() {
                 if (this.isTerminated && e instanceof InterruptedException){
                     return;
                 } else {
+                    e.printStackTrace();
                     System.err.println("SendLogCacheConsumer[" + Thread.currentThread().getName() + "] occurred exception [" + e.getLocalizedMessage() + "]");
                    // For the unknown exception clear the cache
                    sendBuffer.clear();

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/http/Json.java

@@ -91,6 +91,24 @@ public static String toJson(Object obj, Class<?> model) {
         return null;
     }
 
+    public static String toJson(Object obj, Class<?> model, boolean beautify){
+        ObjectWriter writer = mapper.writer();
+        if(null != obj){
+            try{
+                if(null != model){
+                    writer = writer.withView(model);
+                }
+                if(beautify){
+                    return writer.withDefaultPrettyPrinter().writeValueAsString(obj);
+                }
+                return writer.writeValueAsString(obj);
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        return null;
+    }
+
     /**
      * Convert object using serialization and deserialization
      *

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisConfigAutowired.java → streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/plugin/StreamisConfigAutowired.java

@@ -1,4 +1,4 @@
-package com.webank.wedatasphere.streamis.jobmanager.log.collector;
+package com.webank.wedatasphere.streamis.jobmanager.plugin;
 
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
 
@@ -11,5 +11,5 @@ public interface StreamisConfigAutowired {
      * Log appender config
      * @param builder builder
      */
-    StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Builder builder);
+    StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Builder builder) throws Exception;
 }

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisLogAppenderTest.java

@@ -1,16 +1,28 @@
 package com.webank.wedatasphere.streamis.jobmanager.log.collector;
 
+import org.apache.logging.log4j.LogManager;
+import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class StreamisLogAppenderTest {
     private static final Logger LOG = LoggerFactory.getLogger(StreamisLogAppenderTest.class);
-    public static void main(String[] args) throws InterruptedException {
-        while(true){
-            for(int i = 0; i < 100; i ++){
-                LOG.info("Stream Log appender test");
+    @Test
+    public void appenderLog() throws InterruptedException {
+        org.apache.logging.log4j.core.Logger logger = (org.apache.logging.log4j.core.Logger) LogManager.getLogger("Test: Hello-world");
+        logger.info("");
+        int total = 1000;
+        int tps = 100;
+        long timer = System.currentTimeMillis() + 1000;
+        for(int i = 0; i < total; i ++){
+            if (i > 0 && i % tps == 0){
+                long sleep = timer - System.currentTimeMillis();
+                if (sleep > 0){
+                    Thread.sleep(sleep);
+                }
+                timer = System.currentTimeMillis() + 1000;
             }
-            Thread.sleep(1000);
+            LOG.info("Stream Log appender test, sequence id: " + i);
         }
     }
 }

streamis-jobmanager/streamis-job-log/streamis-job-log-common/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/entities/StreamisLogEvents.java

@@ -21,13 +21,19 @@ public StreamisLogEvents(String applicationName, StreamisLogEvent[] events){
         this.appName = applicationName;
         this.events = events;
         long maxTime = -1;
-        for(StreamisLogEvent event : events){
-            long time = event.getLogTimeStamp();
-            if (time > maxTime){
-                maxTime = time;
+        StreamisLogEvent lastEvent = events[events.length - 1];
+        if (null == lastEvent) {
+            for (StreamisLogEvent event : events) {
+                long time = event.getLogTimeStamp();
+                if (time > maxTime) {
+                    maxTime = time;
+                }
             }
+            this.logTimeInMills = maxTime;
+        }else {
+            this.logTimeInMills = lastEvent.getLogTimeStamp();
         }
-        this.logTimeInMills = maxTime;
+
     }
 
     @Override

streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/conf/JobConf.scala

@@ -15,7 +15,7 @@
 
 package com.webank.wedatasphere.streamis.jobmanager.manager.conf
 
-import org.apache.linkis.common.conf.{CommonVars, TimeType}
+import org.apache.linkis.common.conf.{CommonVars, Configuration, TimeType}
 import com.webank.wedatasphere.streamis.jobmanager.manager.exception.JobExecuteErrorException
 
 
@@ -29,6 +29,21 @@ object JobConf {
 
   val STREAMIS_JOB_PARAM_BLANK_PLACEHOLDER: CommonVars[String] =  CommonVars("wds.streamis.job.param.blank.placeholder", "\u0001")
 
+  /**
+   * Config group for streamis internal configuration
+   */
+  val STREAMIS_INTERNAL_CONFIG_GROUP: CommonVars[String] = CommonVars("wds.streamis.job.internal.config.group", "wds.streamis.internal.params")
+
+  /**
+   * Gateway for stream job log module
+   */
+  val STREAMIS_JOB_LOG_GATEWAY: CommonVars[String] = CommonVars("wds.streamis.job.log.gateway", Configuration.getGateWayURL())
+
+  /**
+   * Path for collecting stream job log
+   */
+  val STREAMIS_JOB_LOG_COLLECT_PATH: CommonVars[String] = CommonVars("wds.streamis.job.log.collect.path", "/api/rest_j/v1/streamis/streamJobManager/log/collect/events")
+
   val FLINK_JOB_STATUS_NOT_STARTED: CommonVars[Int] = CommonVars("wds.streamis.job.status.not-started", 0,"Not Started")
 
   val FLINK_JOB_STATUS_COMPLETED: CommonVars[Int] = CommonVars("wds.streamis.job.status.completed", 1,"Completed")

streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/transform/builder/AbstractStreamisTransformJobBuilder.scala

@@ -18,12 +18,15 @@ package com.webank.wedatasphere.streamis.jobmanager.manager.transform.builder
 import org.apache.linkis.common.conf.CommonVars
 import org.apache.linkis.manager.label.entity.engine.RunType.RunType
 import com.webank.wedatasphere.streamis.jobmanager.launcher.service.StreamJobConfService
+import com.webank.wedatasphere.streamis.jobmanager.manager.conf.JobConf
 import com.webank.wedatasphere.streamis.jobmanager.manager.dao.StreamJobMapper
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamJob
 import com.webank.wedatasphere.streamis.jobmanager.manager.transform.StreamisTransformJobBuilder
 import com.webank.wedatasphere.streamis.jobmanager.manager.transform.entity.{StreamisJobEngineConnImpl, StreamisTransformJob, StreamisTransformJobContent, StreamisTransformJobImpl}
 import org.springframework.beans.factory.annotation.Autowired
 
+import java.util
+import scala.collection.JavaConverters.mapAsJavaMapConverter
 /**
   * Created by enjoyyin on 2021/9/22.
   */
@@ -39,7 +42,12 @@ abstract class AbstractStreamisTransformJobBuilder extends StreamisTransformJobB
   override def build(streamJob: StreamJob): StreamisTransformJob = {
     val transformJob = createStreamisTransformJob()
     transformJob.setStreamJob(streamJob)
-    transformJob.setConfigMap(streamJobConfService.getJobConfig(streamJob.getId))
+    val jobConfig: util.Map[String, Any] = Option(streamJobConfService.getJobConfig(streamJob.getId))
+      .getOrElse(new util.HashMap[String, Any]())
+    // Put and overwrite internal group, users cannot customize the internal configuration
+    jobConfig.put(JobConf.STREAMIS_INTERNAL_CONFIG_GROUP.getValue, new util.HashMap[String, Any]())
+    internalLogConfig(jobConfig)
+    transformJob.setConfigMap(jobConfig)
 //    transformJob.setConfig(configurationService.getFullTree(streamJob.getId))
     val streamJobVersions = streamJobMapper.getJobVersions(streamJob.getId)
     // 无需判断streamJobVersions是否非空，因为TaskService已经判断了
@@ -48,6 +56,14 @@ abstract class AbstractStreamisTransformJobBuilder extends StreamisTransformJobB
     transformJob
   }
 
+  /**
+   * Log internal configuration
+   * @param internal internal config group
+   */
+  private def internalLogConfig(internal: util.Map[String, Any]): Unit = {
+    internal.put(JobConf.STREAMIS_JOB_LOG_GATEWAY.key, JobConf.STREAMIS_JOB_LOG_GATEWAY.getValue)
+    internal.put(JobConf.STREAMIS_JOB_LOG_COLLECT_PATH.key, JobConf.STREAMIS_JOB_LOG_COLLECT_PATH.getValue)
+  }
 }
 
 abstract class AbstractFlinkStreamisTransformJobBuilder extends AbstractStreamisTransformJobBuilder{

streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/transform/impl/FlinkInternalConfigTransform.scala

@@ -0,0 +1,42 @@
+package com.webank.wedatasphere.streamis.jobmanager.manager.transform.impl
+import com.webank.wedatasphere.streamis.jobmanager.launcher.job.LaunchJob
+import com.webank.wedatasphere.streamis.jobmanager.manager.conf.JobConf
+import com.webank.wedatasphere.streamis.jobmanager.manager.transform.impl.FlinkInternalConfigTransform.INTERNAL_CONFIG_MAP
+import org.apache.linkis.common.conf.CommonVars
+
+import java.util
+import scala.collection.JavaConverters.{mapAsJavaMapConverter, mapAsScalaMapConverter}
+
+/**
+ * Flink internal config transform
+ */
+class FlinkInternalConfigTransform extends FlinkConfigTransform {
+
+  /**
+   * Config group name
+   *
+   * @return
+   */
+  override protected def configGroup(): String = JobConf.STREAMIS_INTERNAL_CONFIG_GROUP.getValue
+
+  override protected def transform(internalConfig: util.Map[String, Any], job: LaunchJob): LaunchJob = {
+    transformConfig(internalConfig.asScala.map{
+      case (key, value) =>
+        (FlinkConfigTransform.FLINK_CONFIG_PREFIX + (INTERNAL_CONFIG_MAP.get(key) match {
+          case Some(mappingKey) => mappingKey
+          case _ => value
+        }), value)
+    }.asJava, job)
+  }
+}
+
+object FlinkInternalConfigTransform {
+
+  private val FLINK_LOG_GATEWAY_CONFIG_NAME = CommonVars("wds.streamis.flink.config.name.log-gateway", "stream.log.gateway.address").getValue
+
+  private val FLINK_LOG_COLLECT_PATH_CONFIG_NAME = CommonVars("wds.streamis.flink.config.name.log-collect-path", "stream.log.collect.path").getValue
+
+  val INTERNAL_CONFIG_MAP = Map(JobConf.STREAMIS_JOB_LOG_GATEWAY.key -> FLINK_LOG_GATEWAY_CONFIG_NAME,
+    JobConf.STREAMIS_JOB_LOG_COLLECT_PATH.key -> FLINK_LOG_COLLECT_PATH_CONFIG_NAME
+  )
+}