Extract organization field to ElasticSearch.
Obtained by the @domain part of committer email.
Change-Id: Ib5b3b9996ec78780007cd6f95a47e2dccd570364
diff --git a/kibana/organizations-visualizations.json b/kibana/organizations-visualizations.json
new file mode 100644
index 0000000..c6a3752
--- /dev/null
+++ b/kibana/organizations-visualizations.json
@@ -0,0 +1,30 @@
+[
+ {
+ "_id": "6d3a02f0-8b37-11e7-b340-5faa4252a25c",
+ "_type": "visualization",
+ "_source": {
+ "title": "Organization Pie",
+ "visState": "{\"title\":\"Organization Pie\",\"type\":\"pie\",\"params\":{\"addTooltip\":true,\"addLegend\":true,\"legendPosition\":\"right\",\"isDonut\":false},\"aggs\":[{\"id\":\"1\",\"enabled\":true,\"type\":\"count\",\"schema\":\"metric\",\"params\":{}},{\"id\":\"2\",\"enabled\":true,\"type\":\"terms\",\"schema\":\"segment\",\"params\":{\"field\":\"organization.keyword\",\"size\":500,\"order\":\"desc\",\"orderBy\":\"1\"}}],\"listeners\":{}}",
+ "uiStateJSON": "{}",
+ "description": "",
+ "version": 1,
+ "kibanaSavedObjectMeta": {
+ "searchSourceJSON": "{\"index\":\"gerrit\",\"query\":{\"query_string\":{\"query\":\"*\",\"analyze_wildcard\":true}},\"filter\":[]}"
+ }
+ }
+ },
+ {
+ "_id": "d0c09180-8b38-11e7-b340-5faa4252a25c",
+ "_type": "visualization",
+ "_source": {
+ "title": "Organization Table",
+ "visState": "{\"title\":\"Organization Table\",\"type\":\"table\",\"params\":{\"perPage\":10,\"showPartialRows\":false,\"showMeticsAtAllLevels\":false,\"sort\":{\"columnIndex\":null,\"direction\":null},\"showTotal\":false,\"totalFunc\":\"sum\"},\"aggs\":[{\"id\":\"1\",\"enabled\":true,\"type\":\"sum\",\"schema\":\"metric\",\"params\":{\"field\":\"num_commits\",\"customLabel\":\"# Commits\"}},{\"id\":\"2\",\"enabled\":true,\"type\":\"terms\",\"schema\":\"bucket\",\"params\":{\"field\":\"organization.keyword\",\"size\":10,\"order\":\"desc\",\"orderBy\":\"1\"}},{\"id\":\"3\",\"enabled\":true,\"type\":\"cardinality\",\"schema\":\"metric\",\"params\":{\"field\":\"project.keyword\",\"customLabel\":\"# Projects\"}},{\"id\":\"4\",\"enabled\":true,\"type\":\"cardinality\",\"schema\":\"metric\",\"params\":{\"field\":\"email.keyword\",\"customLabel\":\"# Authors\"}}],\"listeners\":{}}",
+ "uiStateJSON": "{\"vis\":{\"params\":{\"sort\":{\"columnIndex\":null,\"direction\":null}}}}",
+ "description": "",
+ "version": 1,
+ "kibanaSavedObjectMeta": {
+ "searchSourceJSON": "{\"index\":\"gerrit\",\"query\":{\"query_string\":{\"query\":\"*\",\"analyze_wildcard\":true}},\"filter\":[]}"
+ }
+ }
+ }
+]
\ No newline at end of file
diff --git a/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTrasformations.scala b/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTrasformations.scala
index 8756178..01e6dc5 100644
--- a/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTrasformations.scala
+++ b/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTrasformations.scala
@@ -3,8 +3,8 @@
import java.io.{BufferedReader, IOException, InputStreamReader}
import java.net.URL
import java.nio.charset.StandardCharsets
-import java.time.{LocalDateTime, ZoneId, ZoneOffset, ZonedDateTime}
import java.time.format.DateTimeFormatter
+import java.time.{LocalDateTime, ZoneId, ZoneOffset, ZonedDateTime}
import com.gerritforge.analytics.model.{GerritEndpointConfig, ProjectContribution, ProjectContributionSource}
import org.apache.spark.rdd.RDD
@@ -23,6 +23,11 @@
ZoneOffset.UTC, ZoneId.of("Z")
) format DateTimeFormatter.ISO_OFFSET_DATE_TIME
+ private[analytics] val emailToDomain: PartialFunction[JValue,String] = {
+ case JString(email) if email.contains("@") && !email.endsWith("@") =>
+ email.split("@").last.toLowerCase
+ }
+
private[analytics] def transformLongDateToISO(in: String): JObject = {
parse(in).transformField {
case JField(fieldName, JInt(v)) if (fieldName=="date" || fieldName=="last_commit_date") =>
@@ -30,12 +35,19 @@
}.asInstanceOf[JObject]
}
+ private[analytics] def transformAddOrganization(in: JObject): JObject = {
+ Some(in \ "email")
+ .collect(emailToDomain)
+ .fold(in)(org => ("organization" -> org) ~ in)
+ }
+
def getFileContentAsProjectContributions(sourceUrl: String, projectName: String): Iterator[ProjectContribution] = {
val is = new URL(sourceUrl).openConnection.getInputStream
new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))
.lines.iterator().asScala
.filterNot(_.trim.isEmpty)
.map(transformLongDateToISO)
+ .map(transformAddOrganization)
.map(ProjectContribution(projectName, _))
}
@@ -74,6 +86,3 @@
}
}
-
-
-
diff --git a/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala b/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
index e26013e..b846df6 100644
--- a/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
+++ b/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
@@ -6,9 +6,9 @@
import org.json4s.JsonDSL._
import org.json4s._
import org.json4s.jackson.JsonMethods._
-import org.scalatest.{FlatSpec, Inside, Matchers}
+import org.scalatest.{FlatSpec, Inside, Matchers, PartialFunctionValues}
-class GerritAnalyticsTransformationsSpec extends FlatSpec with Matchers with SparkTestSupport with Inside {
+class GerritAnalyticsTransformationsSpec extends FlatSpec with Matchers with SparkTestSupport with Inside with PartialFunctionValues {
import com.gerritforge.analytics.engine.GerritAnalyticsTrasformations._
@@ -30,21 +30,6 @@
}
}
- it should "return one line enriched with its contributor" in {
- val contributorsJson: JObject = ("foo" -> "bar")
- val projectSource = ProjectContributionSource("project", newSource(contributorsJson))
-
- withSparkContext { sc =>
- val projectContributions = sc.parallelize(Seq(projectSource))
- .fetchContributors
- .collect()
-
- projectContributions should contain(
- ProjectContribution("project", contributorsJson)
- )
- }
- }
-
it should "return two lines enriched with its two contributors" in {
val contributor1: JObject = ("foo" -> "bar")
val contributor2: JObject = ("first" -> "John")
@@ -66,16 +51,17 @@
val pc1 = ProjectContribution("project", ("foo" -> "bar"))
val pc2 = ProjectContribution("project", ("foo2" -> "bar2"))
withSparkContext { sc =>
- sc.parallelize(Seq(pc1,pc2)).toJson.collect should contain allOf(
- """{"project":"project","foo":"bar"}""",
- """{"project":"project","foo2":"bar2"}""")
+ sc.parallelize(Seq(pc1, pc2)).toJson.collect should contain allOf(
+ """{"project":"project","foo":"bar"}""",
+ """{"project":"project","foo2":"bar2"}""")
}
}
+
"getFileContentAsProjectContributions" should "collect contributors and handle utf-8" in {
val contributor1: JObject = ("foo" -> "bar")
- val contributor2: JObject = ("first" -> "A with macron in unicode is: \u0100")
+ val contributor2: JObject = ("first" -> "(A with macron) in unicode is: \u0100")
val url = newSource(contributor1, contributor2)
- val projectContributions = getFileContentAsProjectContributions(url,"project").toArray
+ val projectContributions = getFileContentAsProjectContributions(url, "project").toArray
projectContributions should contain allOf(
ProjectContribution("project", contributor1),
@@ -83,6 +69,16 @@
)
}
+ "emailToDomain" should "parse domain" in {
+ emailToDomain.valueAt("a@x.y") should be ("x.y")
+ }
+
+ "transformAddOrganization" should "add organization" in {
+ val contributor: JObject = ("name" -> "contributor1") ~ ("email" -> "name1@domain1")
+ val transformed = transformAddOrganization(contributor)
+ transformed \ "organization" should be(JString("domain1"))
+ }
+
private def newSource(contributorsJson: JObject*): String = {
val tmpFile = File.createTempFile(System.getProperty("java.io.tmpdir"),
s"${getClass.getName}-${System.nanoTime()}")
