Create separate package for date and time operations - Move date time operations into its own package - Test date time opeartions - Create reusable and tested date time formatters Bug: Issue 8833 Change-Id: Iba2581bee0819d5c446c2ddf279f5d9bc3a500a9

diff --git a/src/main/scala/com/gerritforge/analytics/engine/events/AggregationStrategy.scala b/src/main/scala/com/gerritforge/analytics/engine/events/AggregationStrategy.scala
index 9b5990a..95aa113 100644
--- a/src/main/scala/com/gerritforge/analytics/engine/events/AggregationStrategy.scala
+++ b/src/main/scala/com/gerritforge/analytics/engine/events/AggregationStrategy.scala

@@ -14,8 +14,10 @@
 
 package com.gerritforge.analytics.engine.events
 
-import java.text.{DateFormat, SimpleDateFormat}
-import java.time.{LocalDateTime, ZoneOffset}
+import java.text.DateFormat
+import java.time.LocalDateTime
+
+import com.gerritforge.analytics.support.ops.AnalyticsTimeOps.{AnalyticsDateTimeFormater, CommonTimeOperations}
 
 import scala.util.Try
 
@@ -25,7 +27,7 @@
   case class DateTimeParts(year: Integer, month: Integer, day: Integer, hour: Integer)
 
   def decomposeTimeOfAggregatedEvent(event: GerritJsonEvent): DateTimeParts = {
-    val date = LocalDateTime.ofEpochSecond(event.eventCreatedOn, 0, ZoneOffset.UTC)
+    val date: LocalDateTime = CommonTimeOperations.utcDateTimeFromEpoch(event.eventCreatedOn)
     DateTimeParts(date.getYear, date.getMonthValue, date.getDayOfMonth, date.getHour)
   }
 }
@@ -58,27 +60,28 @@
   }
 
   object aggregateByEmailAndHour extends EmailAndTimeBasedAggregation {
-    val dateFormat = new SimpleDateFormat("yyyyMMddHH")
+    val dateFormat = AnalyticsDateTimeFormater.yyyyMMddHH
   }
 
   object aggregateByEmailAndDay extends EmailAndTimeBasedAggregation {
-    val dateFormat = new SimpleDateFormat("yyyyMMdd")
+    val dateFormat = AnalyticsDateTimeFormater.yyyyMMdd
 
     override def decomposeTimeOfAggregatedEvent(event: GerritJsonEvent): DateTimeParts =
       super.decomposeTimeOfAggregatedEvent(event).copy(hour = 0)
   }
 
   object aggregateByEmailAndMonth extends EmailAndTimeBasedAggregation {
-    val dateFormat = new SimpleDateFormat("yyyyMM")
+    val dateFormat = AnalyticsDateTimeFormater.yyyyMM
 
     override def decomposeTimeOfAggregatedEvent(event: GerritJsonEvent): DateTimeParts =
       super.decomposeTimeOfAggregatedEvent(event).copy(day = 0, hour = 0)
   }
 
   object aggregateByEmailAndYear extends EmailAndTimeBasedAggregation {
-    val dateFormat = new SimpleDateFormat("yyyy")
+    val dateFormat = AnalyticsDateTimeFormater.yyyy
 
     override def decomposeTimeOfAggregatedEvent(event: GerritJsonEvent): DateTimeParts =
       super.decomposeTimeOfAggregatedEvent(event).copy(month = 0, day = 0, hour = 0)
   }
+
 }
\ No newline at end of file

diff --git a/src/main/scala/com/gerritforge/analytics/job/Main.scala b/src/main/scala/com/gerritforge/analytics/job/Main.scala
index dc7be1a..3d1c4d4 100644
--- a/src/main/scala/com/gerritforge/analytics/job/Main.scala
+++ b/src/main/scala/com/gerritforge/analytics/job/Main.scala

@@ -14,12 +14,13 @@
 
 package com.gerritforge.analytics.job
 
-import java.time.format.DateTimeFormatter
-import java.time.{LocalDate, ZoneId}
+import java.time.LocalDate
 
 import com.gerritforge.analytics.engine.events.GerritEventsTransformations.NotParsableJsonEvent
 import com.gerritforge.analytics.engine.events.{AggregationStrategy, EventParser, GerritJsonEvent}
 import com.gerritforge.analytics.model.{GerritEndpointConfig, GerritProjectsSupport}
+import com.gerritforge.analytics.support.ops.AnalyticsTimeOps
+import com.gerritforge.analytics.support.ops.AnalyticsTimeOps.AnalyticsDateTimeFormater
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
@@ -41,13 +42,14 @@
     }
   }
 
-  implicit val localDateRead: Read[LocalDate] = reads { str =>
-    val format = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.of("UTC"))
+  implicit val localDateRead: Read[LocalDate] = reads { dateStr =>
+    val cliDateFormat = AnalyticsDateTimeFormater.yyyy_MM_dd
     try {
-      LocalDate.parse(str, format)
+      import AnalyticsTimeOps.implicits._
+      dateStr.parseStringToLocalDate(cliDateFormat).get
     } catch {
       case NonFatal(e) =>
-        throw new IllegalArgumentException(s"Invalid date '$str' expected format is '$format'")
+        throw new IllegalArgumentException(s"Invalid date '$dateStr' expected format is '${cliDateFormat}'")
     }
   }
 
@@ -81,7 +83,7 @@
     opt[String]("events") optional() action { (eventsPath, config) =>
       config.copy(eventsPath = Some(eventsPath))
     } text "location where to load the Gerrit Events"
-    opt[String]("writeNotProcessedEventsTo") optional() action{ (failedEventsPath, config) =>
+    opt[String]("writeNotProcessedEventsTo") optional() action { (failedEventsPath, config) =>
       config.copy(eventsFailureOutputPath = Some(failedEventsPath))
     } text "location where to write a TSV file containing the events we couldn't process with a description fo the reason why"
   }
@@ -107,7 +109,8 @@
   }
 }
 
-trait Job { self: LazyLogging =>
+trait Job {
+  self: LazyLogging =>
   implicit val codec = Codec.ISO8859
 
   def buildProjectStats()(implicit config: GerritEndpointConfig, spark: SparkSession): DataFrame = {
@@ -128,7 +131,7 @@
 
     val projects = GerritProjectsSupport.parseJsonProjectListResponse(Source.fromURL(config.gerritProjectsUrl))
 
-    logger.info(s"Loaded a list of ${projects.size} projects ${if(projects.size > 20) projects.take(20).mkString("[", ",", ", ...]") else projects.mkString("[", ",", "]")}")
+    logger.info(s"Loaded a list of ${projects.size} projects ${if (projects.size > 20) projects.take(20).mkString("[", ",", ", ...]") else projects.mkString("[", ",", "]")}")
 
     val aliasesDF = getAliasDF(config.emailAlias)
 
@@ -136,7 +139,7 @@
 
     val failedEvents: RDD[NotParsableJsonEvent] = events.collect { case Left(eventFailureDescription) => eventFailureDescription }
 
-    if(!failedEvents.isEmpty()) {
+    if (!failedEvents.isEmpty()) {
       config.eventsFailureOutputPath.foreach { failurePath =>
         logger.info(s"Events failures will be stored at '$failurePath'")
 
@@ -148,11 +151,11 @@
     //We might want to use the time of the events as information to feed to the collection of data from the repository
     val repositoryAlteringEvents = events.collect { case Right(event) => event }.repositoryWithNewRevisionEvents
 
-    val firstEventDateMaybe: Option[LocalDate] = if(repositoryAlteringEvents.isEmpty()) None else Some(repositoryAlteringEvents.earliestEventTime.toLocalDate)
+    val firstEventDateMaybe: Option[LocalDate] = if (repositoryAlteringEvents.isEmpty()) None else Some(repositoryAlteringEvents.earliestEventTime.toLocalDate)
 
     val configWithOverriddenUntil = firstEventDateMaybe.fold(config) { firstEventDate =>
       val lastAggregationDate = firstEventDate.plusMonths(1)
-      if(lastAggregationDate.isBefore(LocalDate.now())) {
+      if (lastAggregationDate.isBefore(LocalDate.now())) {
         logger.info(s"Overriding 'until' date '${config.until}' with '$lastAggregationDate' since events ara available until $firstEventDate")
         config.copy(until = Some(lastAggregationDate))
       } else {
@@ -167,17 +170,17 @@
 
     require(statsFromAnalyticsPlugin.schema == statsFromEvents.schema,
       s""" Schemas from the stats collected from events and from the analytics datasets differs!!
-        | From analytics plugin: ${statsFromAnalyticsPlugin.schema}
-        | From gerrit events: ${statsFromEvents.schema}
+         | From analytics plugin: ${statsFromAnalyticsPlugin.schema}
+         | From gerrit events: ${statsFromEvents.schema}
       """.stripMargin)
 
     (statsFromAnalyticsPlugin union statsFromEvents).dashboardStats(aliasesDF)
   }
 
-  def loadEvents(implicit config: GerritEndpointConfig, spark: SparkSession): RDD[Either[NotParsableJsonEvent,GerritJsonEvent]] = { // toDF
+  def loadEvents(implicit config: GerritEndpointConfig, spark: SparkSession): RDD[Either[NotParsableJsonEvent, GerritJsonEvent]] = { // toDF
     import com.gerritforge.analytics.engine.events.GerritEventsTransformations._
 
-    config.eventsPath.fold(spark.sparkContext.emptyRDD[Either[NotParsableJsonEvent,GerritJsonEvent]]) { eventsPath =>
+    config.eventsPath.fold(spark.sparkContext.emptyRDD[Either[NotParsableJsonEvent, GerritJsonEvent]]) { eventsPath =>
       spark
         .read.textFile(eventsPath).rdd
         .parseEvents(EventParser)

diff --git a/src/main/scala/com/gerritforge/analytics/model/GerritEndpointConfig.scala b/src/main/scala/com/gerritforge/analytics/model/GerritEndpointConfig.scala
index 899e56a..c3ea2be 100644
--- a/src/main/scala/com/gerritforge/analytics/model/GerritEndpointConfig.scala
+++ b/src/main/scala/com/gerritforge/analytics/model/GerritEndpointConfig.scala

@@ -15,7 +15,9 @@
 package com.gerritforge.analytics.model
 
 import java.time.format.DateTimeFormatter
-import java.time.{LocalDate, ZoneId}
+import java.time.{LocalDate, ZoneOffset}
+
+import com.gerritforge.analytics.support.ops.AnalyticsTimeOps.AnalyticsDateTimeFormater
 
 case class GerritEndpointConfig(baseUrl: String = "",
                                 prefix: Option[String] = None,
@@ -38,7 +40,7 @@
   }
 
   @transient
-  private lazy val format = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.of("UTC"))
+  private lazy val format: DateTimeFormatter = AnalyticsDateTimeFormater.yyyy_MM_dd.withZone(ZoneOffset.UTC)
   val queryString = Seq("since" -> since.map(format.format), "until" -> until.map(format.format), "aggregate" -> aggregate)
     .flatMap(queryOpt).mkString("?", "&", "")
 

diff --git a/src/main/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOps.scala b/src/main/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOps.scala
new file mode 100644
index 0000000..f99697f
--- /dev/null
+++ b/src/main/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOps.scala

@@ -0,0 +1,70 @@
+// Copyright (C) 2018 GerritForge Ltd
+//
+// Licensed 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 com.gerritforge.analytics.support.ops
+
+import java.time.format.DateTimeFormatter
+import java.time.{LocalDateTime, ZoneOffset}
+
+package AnalyticsTimeOps {
+
+  import java.sql.Timestamp
+  import java.text.SimpleDateFormat
+  import java.time.{Instant, LocalDate, OffsetDateTime}
+
+  import scala.util.Try
+
+
+  object AnalyticsDateTimeFormater {
+
+    val yyyy_MM_dd_HHmmss_SSSSSSSSS: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSSS")
+    val yyyy_MM_dd: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
+
+    val yyyyMMddHH: SimpleDateFormat = new SimpleDateFormat("yyyyMMddHH")
+    val yyyyMMdd: SimpleDateFormat = new SimpleDateFormat("yyyyMMdd")
+    val yyyyMM: SimpleDateFormat = new SimpleDateFormat("yyyyMM")
+    val yyyy: SimpleDateFormat = new SimpleDateFormat("yyyy")
+  }
+
+  object CommonTimeOperations {
+    def nowEpoch: Long = Instant.now().getEpochSecond
+
+    def epochToSqlTimestampOps(epoch: Long) = new Timestamp(epoch)
+
+    def nowSqlTimestmap: Timestamp = epochToSqlTimestampOps(nowEpoch)
+
+    def utcDateTimeFromEpoch(epoch: Long): LocalDateTime = LocalDateTime.ofEpochSecond(epoch, 0, ZoneOffset.UTC)
+  }
+
+  object implicits {
+
+    implicit class LocalDateTimeOps(val localDateTime: LocalDateTime) extends AnyVal {
+      def convertToUTCEpochMillis: Long = localDateTime.atOffset(ZoneOffset.UTC).toInstant.toEpochMilli
+
+      def convertToUTCLocalDateTime: OffsetDateTime = localDateTime.atOffset(ZoneOffset.UTC)
+    }
+
+
+    implicit class StringToTimeParsingOps(val dateStr: String) extends AnyVal {
+      def parseStringToUTCEpoch(stringFormat: DateTimeFormatter): Option[Long] =
+        Try(LocalDateTime.parse(dateStr, stringFormat).convertToUTCEpochMillis).toOption
+
+      def parseStringToLocalDate(stringFormat: DateTimeFormatter): Option[LocalDate] =
+        Try(LocalDate.parse(dateStr, stringFormat)).toOption
+    }
+
+  }
+
+}
\ No newline at end of file

diff --git a/src/test/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOpsSpec.scala b/src/test/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOpsSpec.scala
new file mode 100644
index 0000000..c76efdb
--- /dev/null
+++ b/src/test/scala/com/gerritforge/analytics/support/ops/AnalyticsTimeOpsSpec.scala

@@ -0,0 +1,98 @@
+// Copyright (C) 2018 GerritForge Ltd
+//
+// Licensed 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 com.gerritforge.analytics.support.ops
+
+import java.time.{LocalDate, LocalDateTime, ZoneOffset}
+
+import com.gerritforge.analytics.support.ops.AnalyticsTimeOps.AnalyticsDateTimeFormater
+import org.scalatest.{FlatSpec, Matchers}
+
+class AnalyticsTimeOpsSpec extends FlatSpec with Matchers {
+
+
+  "String parser - Given a correct string and date format" should "return an epoch value" in {
+    val epochValueUTC =
+      LocalDateTime
+        .of(2018, 1, 1, 12, 0, 0, 0)
+        .atOffset(ZoneOffset.UTC)
+        .toInstant
+        .toEpochMilli
+
+    val stringDate = "2018-01-01 12:00:00.000000000"
+    val dateFormat = AnalyticsDateTimeFormater.yyyy_MM_dd_HHmmss_SSSSSSSSS
+
+    import AnalyticsTimeOps.implicits._
+
+    stringDate.parseStringToUTCEpoch(dateFormat).get should equal(epochValueUTC)
+  }
+  "String parser - Given a correct string and date format" should "return also a local date" in {
+    val utcLocalDate: LocalDate =
+      LocalDate.of(2018, 1, 1)
+
+    val stringDate = "2018-01-01"
+    val dateFormat = AnalyticsDateTimeFormater.yyyy_MM_dd
+
+    import AnalyticsTimeOps.implicits._
+
+    stringDate.parseStringToLocalDate(dateFormat).get should equal(utcLocalDate)
+  }
+
+
+  "String parser - An incorrect string a given format" should "return None" in {
+    val stringDate = "2018-01-01 12:00:00.000000000"
+    val dateFormat = AnalyticsDateTimeFormater.yyyy_MM_dd
+
+    import AnalyticsTimeOps.implicits._
+    stringDate.parseStringToUTCEpoch(dateFormat) should equal(None)
+  }
+
+
+  "Simple Date Formats" should "convert to the correct strings" in {
+
+    val epochValueUTC =
+      LocalDateTime
+        .of(2018, 1, 1, 12, 0, 0, 0)
+        .atOffset(ZoneOffset.UTC)
+        .toInstant.toEpochMilli
+
+    val yyyyMMddHHStr = "2018010112"
+    val yyyyMMddStr = "20180101"
+    val yyyyMMStr = "201801"
+    val yyyyStr = "2018"
+
+    AnalyticsDateTimeFormater.yyyyMMddHH.format(epochValueUTC) should equal(yyyyMMddHHStr)
+    AnalyticsDateTimeFormater.yyyyMMdd.format(epochValueUTC) should equal(yyyyMMddStr)
+    AnalyticsDateTimeFormater.yyyyMM.format(epochValueUTC) should equal(yyyyMMStr)
+    AnalyticsDateTimeFormater.yyyy.format(epochValueUTC) should equal(yyyyStr)
+  }
+
+
+  "UTC conversion" should "check date operations return always UTC" in {
+    val dateTime =
+      LocalDateTime
+        .of(2018, 1, 1, 12, 0, 0, 0)
+
+    val etcDateTime = dateTime.atOffset(ZoneOffset.ofHours(9))
+    val utcDateTime = dateTime.atOffset(ZoneOffset.UTC)
+
+    import AnalyticsTimeOps.implicits._
+    dateTime.convertToUTCEpochMillis should equal(utcDateTime.toInstant.toEpochMilli)
+    dateTime.convertToUTCEpochMillis should not equal (etcDateTime.toInstant.toEpochMilli)
+
+    dateTime.convertToUTCLocalDateTime should equal(utcDateTime)
+    dateTime.convertToUTCLocalDateTime should not equal (etcDateTime)
+
+  }
+}