ARGO-3977 Improve logs generated by the flink jobs

Author: cthermolia-grnet

flink_jobs_v2/ams_ingest_metric/src/main/java/argo/streaming/AmsIngestMetric.java

@@ -35,6 +35,8 @@
 
 import argo.avro.MetricData;
 import argo.avro.MetricDataOld;
+import org.apache.flink.api.common.JobID;
+import org.slf4j.MDC;
 
 /**
  * Flink Job : Stream metric data from ARGO messaging to Hbase job required cli
@@ -59,6 +61,7 @@ public class AmsIngestMetric {
 
     static Logger LOG = LoggerFactory.getLogger(AmsIngestMetric.class);
     private static String runDate;
+ 
     /**
      * Check if flink job has been called with ams rate params
      */
@@ -109,6 +112,7 @@ public static boolean hasArgs(String[] reqArgs, ParameterTool paramTool) {
 
     public static void main(String[] args) throws Exception {
 
+        configJID();
         // Create flink execution environment
         StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
         see.setParallelism(1);
@@ -138,6 +142,7 @@ public static void main(String[] args) throws Exception {
         if (runDate != null) {
             runDate = runDate + "T00:00:00.000Z";
         }
+              
 
         // Check if checkpointing is desired
         if (hasCheckArgs(parameterTool)) {
@@ -253,5 +258,12 @@ public void flatMap(String value, Collector<MetricData> out) throws Exception {
 
         see.execute(jobTitleSB.toString());
     }
+    private static String getJID() {
+        return JobID.generate().toString();
+    }
 
+    private static void configJID() { //config the JID in the log4j.properties
+        String jobId = getJID();
+        MDC.put("JID", jobId);
+    }
 }

flink_jobs_v2/ams_ingest_metric/src/main/resources/log4j.properties

@@ -20,4 +20,4 @@ log4j.rootLogger=INFO, console
 
 log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
+log4j.appender.console.layout.ConversionPattern= %d{HH:mm:ss,SSS} %X{JID} %-5p %-60c %x - %m%n

flink_jobs_v2/ams_ingest_sync/src/main/java/argo/streaming/AmsIngestSync.java

@@ -2,6 +2,7 @@
 
 import ams.connector.ArgoMessagingSource;
 import java.util.concurrent.TimeUnit;
+import org.apache.flink.api.common.JobID;
 
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
@@ -10,6 +11,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 
 /**
  * Flink Streaming JOB for Ingesting Sync Data to HDFS job required cli
@@ -30,7 +32,6 @@ public class AmsIngestSync {
     // setup logger
     static Logger LOG = LoggerFactory.getLogger(AmsIngestSync.class);
     private static String runDate;
-
     /**
      * Check if a list of expected cli arguments have been provided to this
      * flink job
@@ -56,6 +57,7 @@ public static boolean hasAmsRateArgs(ParameterTool paramTool) {
 
     // main job function
     public static void main(String[] args) throws Exception {
+        configJID();
 
         // Create flink execution enviroment
         StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -76,6 +78,8 @@ public static void main(String[] args) throws Exception {
         if (runDate != null) {
             runDate = runDate + "T00:00:00.000Z";
         }
+           
+ 
 
         // set ams client batch and interval to default values
         int batch = 1;
@@ -118,5 +122,12 @@ public static void main(String[] args) throws Exception {
         see.execute(jobTitleSB.toString());
 
     }
+    private static String getJID() {
+        return JobID.generate().toString();
+    }
 
+    private static void configJID() {
+        String jobId = getJID();
+        MDC.put("JID", jobId);
+    }
 }

flink_jobs_v2/ams_ingest_sync/src/main/java/argo/streaming/SyncHDFSOutputFormat.java

@@ -44,12 +44,13 @@ class SyncHDFSOutputFormat implements OutputFormat<String> {
 
 	// setup logger
 	static Logger LOG = LoggerFactory.getLogger(SyncHDFSOutputFormat.class);
-
+     
 	private static final long serialVersionUID = 1L;
 
 	private URI basePath;
 	private org.apache.hadoop.conf.Configuration hadoopConf;
 	private FileSystem hdfs;
+        
 
 	public void setBasePath(String url) throws URISyntaxException {
 		this.basePath = new URI(url);

flink_jobs_v2/ams_ingest_sync/src/main/resources/log4j.properties

@@ -0,0 +1,23 @@
+################################################################################
+#  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.
+################################################################################
+
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern= %d{HH:mm:ss,SSS} %X{JID} %-5p %-60c %x - %m%n

flink_jobs_v2/batch_multi/src/main/java/argo/batch/ArgoMultiJob.java

@@ -51,8 +51,10 @@
 import trends.status.ServiceTrendsCounter;
 
 import java.util.List;
+import org.apache.flink.api.common.JobID;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
+import org.slf4j.MDC;
 
 /**
  * Implements an ARGO Status Batch Job in flink
@@ -84,7 +86,8 @@
  */
 public class ArgoMultiJob {
 
-    static Logger LOG = LoggerFactory.getLogger(ArgoMultiJob.class);
+    static Logger LOG = LoggerFactory.getLogger( ArgoMultiJob.class);
+
     private static String dbURI;
     private static String reportID;
     private static Integer rankNum;
@@ -97,9 +100,10 @@ public class ArgoMultiJob {
     private static boolean calcStatusTrends = false;
     private static boolean calcFlipFlops = false;
     private static boolean calcTagTrends = false;
-
+  
     public static void main(String[] args) throws Exception {
 
+        configJID();
         final ParameterTool params = ParameterTool.fromArgs(args);
 
         // set up the execution environment
@@ -142,6 +146,7 @@ public static void main(String[] args) throws Exception {
         List<String> confData = confDS.collect();
         ReportManager cfgMgr = new ReportManager();
         cfgMgr.loadJsonString(confData);
+
         enableComputations(cfgMgr.activeComputations, params);
 
         DataSource<String> opsDS = env.fromElements(amr.getResourceJSON(ApiResource.OPS));
@@ -226,7 +231,7 @@ public static void main(String[] args) throws Exception {
         DataSet<StatusMetric> stDetailDS = mdataTotalDS.groupBy("group", "service", "hostname", "metric")
                 .sortGroup("timestamp", Order.ASCENDING).reduceGroup(new CalcPrevStatus(params))
                 .withBroadcastSet(mpsDS, "mps").withBroadcastSet(recDS, "rec").withBroadcastSet(opsDS, "ops");
-     
+
         //Create StatusMetricTimeline dataset for endpoints
         DataSet<StatusTimeline> statusMetricTimeline = stDetailDS.groupBy("group", "service", "hostname", "metric").sortGroup("timestamp", Order.ASCENDING)
                 .reduceGroup(new CalcMetricTimeline(params)).withBroadcastSet(mpsDS, "mps").withBroadcastSet(opsDS, "ops")
@@ -466,7 +471,6 @@ private static void enableComputations(ReportManager.ActiveComputations activeCo
         calcStatusTrends = isOFF(params, "calcStatusTrends", activeComputations);
         calcFlipFlops = isOFF(params, "calcFlipFlops", activeComputations);
         calcTagTrends = isOFF(params, "calcTagTrends", activeComputations);
-        System.out.println("calcTagTrends--- "+calcTagTrends);
 
         if (!calcStatus && !calcAR && !calcStatusTrends && !calcFlipFlops && !calcStatusTrends) {
             System.exit(0);
@@ -496,4 +500,14 @@ public static boolean isOFF(ParameterTool params, String paramName, ReportManage
             }
         }
     }
+
+    private static String getJID() {
+         return JobID.generate().toString();
+    }
+
+    private static void configJID() {//config the JID in the log4j.properties
+        String jobId=getJID();
+        MDC.put("JID", jobId);
+         
+    }
 }

flink_jobs_v2/batch_multi/src/main/java/argo/batch/PickEndpoints.java

@@ -37,6 +37,7 @@ public class PickEndpoints extends RichFlatMapFunction<MetricData, StatusMetric>
 
     final ParameterTool params;
 
+    
     public PickEndpoints(ParameterTool params) {
         this.params = params;
     }
@@ -108,6 +109,7 @@ public void open(Configuration parameters) throws IOException, ParseException {
         if (!this.thr.get(0).isEmpty()) {
             this.thrMgr.parseJSON(this.thr.get(0));
         }
+  
     }
 
     @Override

flink_jobs_v2/batch_multi/src/main/java/utils/Utils.java

@@ -74,7 +74,7 @@ public static boolean isPreviousDate(String format, Date nowDate, Date firstDate
         }
     }
 
-    public static boolean checkParameters(ParameterTool params, String... vars) {
+    public static boolean checkParameters(ParameterTool params,String... vars) {
 
         for (String var : vars) {
 

flink_jobs_v2/batch_multi/src/main/resources/log4j.properties

@@ -20,4 +20,5 @@ log4j.rootLogger=INFO, console
 
 log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
+log4j.appender.console.layout.ConversionPattern= %d{HH:mm:ss,SSS} %X{JID}  %-5p    %-60c %x - %m%n 
+

flink_jobs_v2/batch_multi/src/test/java/argo/batch/ArgoMultiJobTest.java

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.operators.Order;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -77,7 +78,7 @@ public class ArgoMultiJobTest {
     private static boolean calcStatusTrends = false;
     private static boolean calcFlipFlops = false;
     private static boolean calcTagTrends = false;
-
+    
     public ArgoMultiJobTest() {
     }
 
@@ -104,7 +105,7 @@ public void testMain() throws Exception {
         List<String> confData = confDS.collect();
         ReportManager cfgMgr = new ReportManager();
         cfgMgr.loadJsonString(confData);
-      
+       
         enableComputations(cfgMgr.activeComputations, params);
         DataSource<String> apsDS = env.fromElements(amr.getResourceJSON(ApiResource.AGGREGATION));
         DataSource<String> opsDS = env.fromElements(amr.getResourceJSON(ApiResource.OPS));