Add config autowired interface to set the log appender params.

Author: Davidhua1996

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

@@ -0,0 +1,15 @@
+package com.webank.wedatasphere.streamis.jobmanager.log.collector;
+
+import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
+
+/**
+ * Streamis config autowired
+ */
+public interface StreamisConfigAutowired {
+
+    /**
+     * Log appender config
+     * @param builder builder
+     */
+    StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Builder builder);
+}

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

@@ -3,7 +3,6 @@
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.cache.LogCache;
 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.config.StreamisLogAppenderConfigBuilder;
 import com.webank.wedatasphere.streamis.jobmanager.log.collector.sender.StreamisRpcLogSender;
 import com.webank.wedatasphere.streamis.jobmanager.log.entities.StreamisLogEvent;
 import org.apache.logging.log4j.core.Filter;
@@ -18,12 +17,10 @@
 import org.apache.logging.log4j.core.layout.PatternLayout;
 
 import java.io.Serializable;
-import java.nio.Buffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.ServiceLoader;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Streamis rpc log appender
@@ -84,16 +81,21 @@ public static StreamisRpcLogAppender createAppender(@PluginAttribute("name") Str
         if (Objects.isNull(layout)){
             layout = PatternLayout.createDefaultLayout();
         }
-        // Search the config builder
-        List<StreamisLogAppenderConfigBuilder> configBuilders = new ArrayList<>();
+        // Search the config autowired class
+        List<StreamisConfigAutowired> configAutowiredEntities = new ArrayList<>();
         StreamisLogAppenderConfig logAppenderConfig = null;
-        ServiceLoader.load(StreamisLogAppenderConfigBuilder.class,
-                StreamisRpcLogAppender.class.getClassLoader()).iterator().forEachRemaining(configBuilders::add);
-        if (!configBuilders.isEmpty()){
-            logAppenderConfig = configBuilders.get(0).build(applicationName, filter, rpcLogSenderConfig);
+        ServiceLoader.load(StreamisConfigAutowired.class,
+                StreamisRpcLogAppender.class.getClassLoader()).iterator().forEachRemaining(configAutowiredEntities::add);
+        StreamisLogAppenderConfig.Builder builder = new StreamisLogAppenderConfig.Builder(applicationName, filter, rpcLogSenderConfig);
+        for (StreamisConfigAutowired autowired : configAutowiredEntities){
+            logAppenderConfig = autowired.logAppenderConfig(builder);
         }
         if (Objects.isNull(logAppenderConfig)){
-            logAppenderConfig = new StreamisLogAppenderConfig(applicationName, filter, rpcLogSenderConfig);
+            logAppenderConfig = builder.build();
+        }
+        applicationName = logAppenderConfig.getApplicationName();
+        if (null == applicationName || applicationName.trim().equals("")){
+            throw new IllegalArgumentException("Application name cannot be empty");
         }
         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

@@ -23,27 +23,38 @@ public StreamisLogAppenderConfig(String applicationName, Filter filter,
         this.senderConfig = null != rpcLogSenderConfig? rpcLogSenderConfig : new RpcLogSenderConfig();
     }
 
+    public static class Builder{
+        private String applicationName;
+
+        private Filter filter;
+
+        private RpcLogSenderConfig rpcLogSenderConfig;
+
+        public Builder(String applicationName, Filter filter,
+                       RpcLogSenderConfig rpcLogSenderConfig){
+            this.applicationName = applicationName;
+            this.filter = filter;
+            this.rpcLogSenderConfig = rpcLogSenderConfig;
+        }
+
+        StreamisLogAppenderConfig.Builder setAppName(){
+            return null;
+        }
+
+        public StreamisLogAppenderConfig build(){
+            return null;
+        }
+    }
     public String getApplicationName() {
         return applicationName;
     }
 
-    public void setApplicationName(String applicationName) {
-        this.applicationName = applicationName;
-    }
-
     public Filter getFilter() {
         return filter;
     }
 
-    public void setFilter(Filter filter) {
-        this.filter = filter;
-    }
-
     public RpcLogSenderConfig getSenderConfig() {
         return senderConfig;
     }
 
-    public void setSenderConfig(RpcLogSenderConfig senderConfig) {
-        this.senderConfig = senderConfig;
-    }
 }

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

@@ -20,8 +20,8 @@
     <appenders>
         <StreamRpcLog name="StreamRpcLog" appName="stream_application">
             <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n"/>
-            <RpcLogSender address="http://127.0.0.1/log/collector" sendRetryCnt="3"
-                          connectionTimeout="3000" socketTimeout="15000" serverRecoveryTimeInSec="5" maxDelayTimeInSec="60">
+            <RpcLogSender sendRetryCnt="3" connectionTimeout="3000"
+                          socketTimeout="15000" serverRecoveryTimeInSec="5" maxDelayTimeInSec="60">
                 <AuthConfig tokenCodeKey="" tokenCode="" tokenUser="" tokenUserKey=""/>
                 <SendLogCache size="200" maxConsumeThread="10"/>
                 <SendBuffer size="50" expireTimeInSec="2"/>

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/resources/log4j2.xml

@@ -10,9 +10,9 @@ public interface JobLogStorage {
 
     /**
      * Create buckets
-     * @param jobName job name
+     * @param appName application name
      * @param bucketConfig bucket config
      * @return config
      */
-    JobLogBucket getOrCreateBucket(String jobName, JobLogBucketConfig bucketConfig);
+    JobLogBucket getOrCreateBucket(String appName, JobLogBucketConfig bucketConfig);
 }