Release 3.7.0 into master

.circleci/config.yml

@@ -1,14 +1,19 @@
 version: 2.1
 jobs:
   analytics-core-build:
-    machine: true
+    machine:
+      image: ubuntu-2004:202008-01
     steps:
     - checkout
     - restore_cache:
         key: dp-dependency-cache-{{ checksum "pom.xml" }}
     - run:
         name: sunbird-analytics-core-build
-        command: mvn scoverage:report
+        command: |
+          export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64
+          export PATH=$JAVA_HOME/bin:$PATH
+          echo $(java -version)
+          mvn scoverage:report
     - save_cache:
         key: dp-dependency-cache-{{ checksum "pom.xml" }}
         paths: ~/.m2
@@ -20,10 +25,11 @@ jobs:
     - run:
         name: sonar
         command: |
+          export JAVA_HOME=/usr/lib/jvm/java-1.11.0-openjdk-amd64
           mvn -X sonar:sonar -Dsonar.projectKey=project-sunbird_sunbird-analytics-core -Dsonar.organization=project-sunbird -Dsonar.exclusions=analytics-core/src/main/scala/org/ekstep/analytics/streaming/** -Dsonar.host.url=https://sonarcloud.io -Dsonar.scala.coverage.reportPaths=/home/circleci/project/target/scoverage.xml
 
 workflows:
   version: 2.1
   workflow:
     jobs:
-    - analytics-core-build
+    - analytics-core-build

.gitignore

@@ -3,3 +3,5 @@ output-*
 dependency-reduced-pom.xml
 **/target
 **/logs
+**/.idea/**
+*.iml

Jenkinsfile

@@ -32,8 +32,9 @@ node('build-slave') {
         }
         stage('Archive artifacts'){
             sh """
-                        mkdir lpa_artifacts
+                        mkdir lpa_core_artifacts
                         cp analytics-job-driver/target/analytics-framework-2.0.jar lpa_core_artifacts
+                        cp analytics-core/lib/scruid*.jar lpa_core_artifacts
                         zip -j lpa_core_artifacts.zip:${artifact_version} lpa_core_artifacts/*
                     """
             archiveArtifacts artifacts: "lpa_core_artifacts.zip:${artifact_version}", fingerprint: true, onlyIfSuccessful: true

analytics-core/pom.xml

@@ -103,8 +103,7 @@
 		<dependency>
 			<groupId>org.apache.httpcomponents</groupId>
 			<artifactId>httpclient</artifactId>
-			<version>4.5.2</version>
-			<scope>provided</scope>
+			<version>4.5.6</version>
 		</dependency>
 		<dependency>
 			<groupId>com.typesafe</groupId>
@@ -248,12 +247,75 @@
 			<artifactId>commons-text</artifactId>
 			<version>1.6</version>
 		</dependency>
+		<!-- https://mvnrepository.com/artifact/com.typesafe.scala-logging/scala-logging -->
+		<dependency>
+			<groupId>com.typesafe.scala-logging</groupId>
+			<artifactId>scala-logging_2.11</artifactId>
+			<version>3.6.0</version>
+		</dependency>
 		<!-- https://mvnrepository.com/artifact/ing.wbaa.druid/scruid -->
 		<dependency>
 			<groupId>ing.wbaa.druid</groupId>
 			<artifactId>scruid_${scala.maj.version}</artifactId>
-			<version>2.3.0</version>
+			<version>2.4.0</version>
+			<scope>system</scope>
+			<systemPath>${project.basedir}/lib/scruid_2.11-2.4.0.jar</systemPath>
+		</dependency>
+		             <!-- Scruid dependency starts -->
+		<!-- https://mvnrepository.com/artifact/io.circe/circe-core -->
+		<dependency>
+			<groupId>io.circe</groupId>
+			<artifactId>circe-core_2.11</artifactId>
+			<version>0.11.2</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/io.circe/circe-parser -->
+		<dependency>
+			<groupId>io.circe</groupId>
+			<artifactId>circe-parser_2.11</artifactId>
+			<version>0.11.2</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/io.circe/circe-generic -->
+		<dependency>
+			<groupId>io.circe</groupId>
+			<artifactId>circe-generic_2.11</artifactId>
+			<version>0.11.2</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.mdedetrich/akka-stream-json -->
+		<dependency>
+			<groupId>org.mdedetrich</groupId>
+			<artifactId>akka-stream-json_2.11</artifactId>
+			<version>0.4.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.mdedetrich/akka-http-json -->
+		<dependency>
+			<groupId>org.mdedetrich</groupId>
+			<artifactId>akka-http-json_2.11</artifactId>
+			<version>0.4.0</version>
 		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.mdedetrich/akka-stream-circe -->
+		<dependency>
+			<groupId>org.mdedetrich</groupId>
+			<artifactId>akka-stream-circe_2.11</artifactId>
+			<version>0.4.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.mdedetrich/akka-http-circe -->
+		<dependency>
+			<groupId>org.mdedetrich</groupId>
+			<artifactId>akka-http-circe_2.11</artifactId>
+			<version>0.4.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/com.typesafe.akka/akka-http -->
+		<dependency>
+			<groupId>com.typesafe.akka</groupId>
+			<artifactId>akka-http_2.11</artifactId>
+			<version>10.1.9</version>
+		</dependency>
+		<dependency>
+			<groupId>io.circe</groupId>
+			<artifactId>circe-java8_2.11</artifactId>
+			<version>0.11.1</version>
+		</dependency>
+		<!-- Scruid dependency ends -->
 		<dependency>
 			<groupId>org.elasticsearch</groupId>
 			<artifactId>elasticsearch-hadoop</artifactId>
@@ -272,6 +334,12 @@
 			<artifactId>postgresql</artifactId>
 			<version>9.1-901.jdbc4</version>
 		</dependency>
+		<dependency>
+			<groupId>io.zonky.test</groupId>
+			<artifactId>embedded-postgres</artifactId>
+			<version>1.2.6</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
@@ -290,6 +358,8 @@
 							<goal>testCompile</goal>
 						</goals>
 						<configuration>
+							<source>8</source>
+							<target>8</target>
 							<args>
 								<arg>-dependencyfile</arg>
 								<arg>${project.build.directory}/.scala_dependencies</arg>

analytics-core/src/main/scala/org/ekstep/analytics/framework/DataFetcher.scala

@@ -7,11 +7,11 @@ import org.apache.spark.streaming.dstream.DStream
 import org.ekstep.analytics.framework.Level.INFO
 import org.ekstep.analytics.framework.exception.DataFetcherException
 import org.ekstep.analytics.framework.fetcher.{AzureDataFetcher, DruidDataFetcher, S3DataFetcher}
-import org.ekstep.analytics.framework.util.{JSONUtils, JobLogger}
+import org.ekstep.analytics.framework.util.{CommonUtil, JSONUtils, JobLogger}
 
 /**
- * @author Santhosh
- */
+  * @author Santhosh
+  */
 object DataFetcher {
 
     implicit val className = "org.ekstep.analytics.framework.DataFetcher"
@@ -33,37 +33,59 @@ object DataFetcher {
                 AzureDataFetcher.getObjectKeys(search.queries.get);
             case "local" =>
                 JobLogger.log("Fetching the batch data from Local file")
-                search.queries.get.map { x => x.file.getOrElse("") }.filterNot { x => x == null };
+                search.queries.get.map { x => x.file.getOrElse(null) }.filterNot { x => x == null };
             case "druid" =>
                 JobLogger.log("Fetching the batch data from Druid")
                 val data = DruidDataFetcher.getDruidData(search.druidQuery.get)
+                // $COVERAGE-OFF$
+                // Disabling scoverage as the below code cannot be covered as DruidDataFetcher is not mockable being an object and embedded druid is not available yet
                 val druidDataList = data.map(f => JSONUtils.deserialize[T](f))
-                return sc.parallelize(druidDataList);
+                return druidDataList
+            // $COVERAGE-ON$
             case _ =>
                 throw new DataFetcherException("Unknown fetcher type found");
         }
+
         if (null == keys || keys.length == 0) {
             return sc.parallelize(Seq[T](), JobContext.parallelization);
         }
         JobLogger.log("Deserializing Input Data", None, INFO);
+        val filteredKeys = search.queries.get.map{q =>
+            getFilteredKeys(q, keys, q.partitions)
+        }.flatMap(f => f)
+
         val isString = mf.runtimeClass.getName.equals("java.lang.String");
-        sc.textFile(keys.mkString(","), JobContext.parallelization).map { line => {
+        val inputEventsCount = fc.inputEventsCount;
+        sc.textFile(filteredKeys.mkString(","), JobContext.parallelization).map { line => {
             try {
+                inputEventsCount.add(1);
                 if (isString) line.asInstanceOf[T] else JSONUtils.deserialize[T](line);
             } catch {
                 case ex: Exception =>
                     JobLogger.log(ex.getMessage, None, INFO);
                     null.asInstanceOf[T]
-                }
             }
+        }
         }.filter { x => x != null };
     }
 
     /**
-     * API to fetch the streaming data given an array of query objects
-     */
+      * API to fetch the streaming data given an array of query objects
+      */
     def fetchStreamData[T](sc: StreamingContext, search: Fetcher)(implicit mf: Manifest[T]): DStream[T] = {
         null;
     }
 
+    def getFilteredKeys(query: Query, keys: Array[String], partitions: Option[List[Int]]): Array[String] = {
+        if (partitions.nonEmpty) {
+            val finalKeys = keys.map{f =>
+                partitions.get.map{p =>
+                    val reg = raw"(\d{4})-(\d{2})-(\d{2})-$p-".r.findFirstIn(f)
+                    if(reg.nonEmpty && f.contains(reg.get)) f else ""
+                }
+            }.flatMap(f => f)
+            finalKeys.filter(f => f.nonEmpty)
+        }
+        else keys
+    }
 }

analytics-core/src/main/scala/org/ekstep/analytics/framework/DataFilter.scala

@@ -133,16 +133,6 @@ object DataFilter {
                     val eventMap = CommonUtil.caseClassToMap(event)
                     CommonUtil.getTimestamp(eventMap.get("$attimestamp").get.asInstanceOf[String]).asInstanceOf[AnyRef];
                 }
-            case "gameId" =>
-                val gid = getBeanProperty(event, "edata.eks.gid");
-                if (null == gid)
-                    getBeanProperty(event, "gdata.id");
-                else
-                    gid;
-            case "genieTag" =>
-                val tags = if(event.isInstanceOf[Event]) CommonUtil.getETags(event.asInstanceOf[Event]).app else getBeanProperty(event, "etags").asInstanceOf[ETags].app;
-                if (tags.isDefined) tags.get else List()
-            case "gameVersion"      => getBeanProperty(event, "gdata.ver");
             case "userId"           => getBeanProperty(event, "uid");
             case "sessionId"        => getBeanProperty(event, "sid");
             case "telemetryVersion" => getBeanProperty(event, "ver");

analytics-core/src/main/scala/org/ekstep/analytics/framework/FrameworkContext.scala

@@ -1,17 +1,27 @@
 package org.ekstep.analytics.framework
 
-import ing.wbaa.druid.DruidConfig
+import ing.wbaa.druid.{DruidConfig, QueryHost}
 import ing.wbaa.druid.client.DruidClient
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.types.StructType
 import org.sunbird.cloud.storage.BaseStorageService
-import org.sunbird.cloud.storage.conf.AppConf
-import org.sunbird.cloud.storage.factory.{StorageConfig, StorageServiceFactory}
+import org.sunbird.cloud.storage.factory.StorageServiceFactory
 
 import scala.collection.mutable.Map
+import org.ekstep.analytics.framework.util.HadoopFileUtil
+import org.apache.spark.util.LongAccumulator
+import org.ekstep.analytics.framework.conf.AppConf
+import org.ekstep.analytics.framework.fetcher.{AkkaHttpClient, AkkaHttpUtil, DruidDataFetcher}
 
 class FrameworkContext {
 
   var dc: DruidClient = null;
+  var drc: DruidClient = null;
   var storageContainers: Map[String, BaseStorageService] = Map();
+  val fileUtil = new HadoopFileUtil();
+
+  var inputEventsCount: LongAccumulator = _
+  var outputEventsCount: LongAccumulator = _
 
   def initialize(storageServices: Option[Array[(String, String, String)]]) {
     dc = DruidConfig.DefaultConfig.client;
@@ -26,15 +36,23 @@ class FrameworkContext {
     getStorageService(storageType, storageType, storageType);
   }
 
+  def getHadoopFileUtil(): HadoopFileUtil = {
+    return fileUtil;
+  }
+
   def getStorageService(storageType: String, storageKey: String, storageSecret: String): BaseStorageService = {
+    if("local".equals(storageType)) {
+      return null;
+    }
     if (!storageContainers.contains(storageType + "|" + storageKey)) {
-      storageContainers.put(storageType, StorageServiceFactory.getStorageService(StorageConfig(storageType, AppConf.getStorageKey(storageKey), AppConf.getStorageSecret(storageSecret))));
+      storageContainers.put(storageType + "|" + storageKey, StorageServiceFactory.getStorageService(org.sunbird.cloud.storage.factory.StorageConfig(storageType, AppConf.getConfig(storageKey), AppConf.getConfig(storageSecret))));
     }
-    storageContainers.get(storageType).get
+    storageContainers.get(storageType + "|" + storageKey).get
   }
 
-  def setDruidClient(druidClient: DruidClient) {
+  def setDruidClient(druidClient: DruidClient, druidRollupClient: DruidClient) {
     dc = druidClient;
+    drc = druidRollupClient;
   }
 
   def getDruidClient(): DruidClient = {
@@ -44,19 +62,49 @@ class FrameworkContext {
     return dc;
   }
 
+  def getDruidRollUpClient(): DruidClient = {
+    if (null == drc) {
+      val conf = DruidConfig.DefaultConfig
+      drc = DruidConfig.apply(
+        Seq(QueryHost(AppConf.getConfig("druid.rollup.host"), AppConf.getConfig("druid.rollup.port").toInt)),
+        conf.secure,
+        conf.url,conf.healthEndpoint,conf.datasource,conf.responseParsingTimeout,conf.clientBackend,
+        conf.clientConfig,conf.scanQueryLegacyMode,conf.zoneId,conf.system).client
+    }
+    return drc;
+  }
+
+  def getAkkaHttpUtil(): AkkaHttpClient = {
+      AkkaHttpUtil
+  }
+
   def shutdownDruidClient() = {
     if (dc != null) dc.actorSystem.terminate()
   }
 
+  def shutdownDruidRollUpClien() = {
+    if (drc != null) drc.actorSystem.terminate()
+  }
+
   def shutdownStorageService() = {
-    if (null != storageContainers) {
+    if (storageContainers.nonEmpty) {
       storageContainers.foreach(f => f._2.closeContext());
     }
   }
 
   def closeContext() = {
     shutdownDruidClient();
+    shutdownDruidRollUpClien();
     shutdownStorageService();
   }
 
+  def loadData(spark: SparkSession, settings: scala.collection.Map[String, String], url: String, schema: StructType): DataFrame = {
+    if (schema.nonEmpty) {
+      spark.read.schema(schema).format(url).options(settings).load()
+    }
+    else {
+      spark.read.format(url).options(settings).load()
+    }
+  }
+
 }