Enrich user info in Auditlogs ETL Allow enrichment of user information. This change on allows "user type" to be defined, but it can be easily extended to any other useful information. User data is passed via a CSV file with the following format: id,type 123,bot 456,human "User type" is defaulted to "human" if not present in this file. A possible use case for this feature is the distinction between "bot" and "human" users. Feature: Issue 10229 Change-Id: I042244154e668f5137ff97053f1cc2ad88342acd
diff --git a/README.md b/README.md index 165c68a..9f5dfa4 100644 --- a/README.md +++ b/README.md
@@ -147,15 +147,25 @@ ## Parameters -* -u, --gerritUrl - gerrit server URL (Required) -* --username - Gerrit API Username (Optional) -* --password - Gerrit API Password (Optional) -* -i, --elasticSearchIndex - elasticSearch index to persist data into (Required) -* -p, --eventsPath - path to a directory (or a file) containing auditLogs events. Supports also _.gz_ files. (Required) -* -a, --eventsTimeAggregation - Events of the same type, produced by the same user will be aggregated with this time granularity: 'second', 'minute', 'hour', 'week', 'month', 'quarter'. (Optional) - Default: 'hour' -* -k, --ignoreSSLCert - Ignore SSL certificate validation (Optional) - Default: false -* -s, --since - process only auditLogs occurred after (and including) this date (Optional) -* -u, --until - process only auditLogs occurred before (and including) this date (Optional) +* -u, --gerritUrl - gerrit server URL (Required) +* --username - Gerrit API Username (Optional) +* --password - Gerrit API Password (Optional) +* -i, --elasticSearchIndex - elasticSearch index to persist data into (Required) +* -p, --eventsPath - path to a directory (or a file) containing auditLogs events. Supports also _.gz_ files. (Required) +* -a, --eventsTimeAggregation - Events of the same type, produced by the same user will be aggregated with this time granularity: 'second', 'minute', 'hour', 'week', 'month', 'quarter'. (Optional) - Default: 'hour' +* -k, --ignoreSSLCert - Ignore SSL certificate validation (Optional) - Default: false +* -s, --since - process only auditLogs occurred after (and including) this date (Optional) +* -u, --until - process only auditLogs occurred before (and including) this date (Optional) +* -a, --additionalUserInfoPath - path to a CSV file containing additional user information (Optional). Currently it is only possible to add user `type` (i.e.: _bot_, _human_). +If the type is not specified the user will be considered _human_. + + Here an additional user information CSV file example: + ```csv + id,type + 123,"bot" + 456,"bot" + 789,"human" + ``` ### Build
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/AdditionalUserInfo.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/AdditionalUserInfo.scala new file mode 100644 index 0000000..4400a70 --- /dev/null +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/AdditionalUserInfo.scala
@@ -0,0 +1,45 @@ +package com.gerritforge.analytics.auditlog.broadcast + +import com.gerritforge.analytics.auditlog.model.AuditLogETLConfig +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.types.{IntegerType, StringType, StructType} + +import scala.util.Try + +case class AdditionalUsersInfo(usersInfo: Map[GerritAccountId,AdditionalUserInfo]) { + def getUserType(who: Int): String = usersInfo.get(who).map(_.`type`).getOrElse(AdditionalUserInfo.DEFAULT_USER_TYPE) +} + +object AdditionalUsersInfo { + val empty: AdditionalUsersInfo = AdditionalUsersInfo(Map.empty[GerritAccountId,AdditionalUserInfo]) +} + +case class AdditionalUserInfo(id: GerritAccountId, `type`: String) + +object AdditionalUserInfo { + val DEFAULT_USER_TYPE = "human" + + def loadAdditionalUserInfo(config: AuditLogETLConfig)(implicit spark: SparkSession): Try[AdditionalUsersInfo] = { + + val schema = new StructType() + .add("id", IntegerType,false) + .add("type", StringType,false) + + import spark.implicits._ + Try { + AdditionalUsersInfo( + config.additionalUserInfoPath.map { path => + spark.read + .option("header", "true") + .schema(schema) + .csv(path) + .as[AdditionalUserInfo] + // We are collecting on the fair assumption that the additional user info file will fit in memory + .collect + .map(additionalUserInfo => additionalUserInfo.id -> additionalUserInfo) + .toMap + }.getOrElse(Map.empty[GerritAccountId,AdditionalUserInfo]) + ) + } + } +}
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/GerritUserIdentifiers.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/GerritUserIdentifiers.scala index 0c805fb..9f040d2 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/GerritUserIdentifiers.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/GerritUserIdentifiers.scala
@@ -25,13 +25,13 @@ import org.json4s.FieldSerializer._ -case class GerritUserIdentifiers(private val accounts: Map[GerriAccountId, GerritUserIdentifier]) { - def getIdentifier(accountId: GerriAccountId): GerritUserIdentifier = accounts.getOrElse(accountId, s"$accountId") +case class GerritUserIdentifiers(private val accounts: Map[GerritAccountId, GerritUserIdentifier]) { + def getIdentifier(accountId: GerritAccountId): GerritUserIdentifier = accounts.getOrElse(accountId, s"$accountId") } object GerritUserIdentifiers extends LazyLogging { - private case class GerritAccount(accountId: GerriAccountId, username: Option[String], email: Option[String], name: Option[String]) { + private case class GerritAccount(accountId: GerritAccountId, username: Option[String], email: Option[String], name: Option[String]) { val getIdentifier: GerritUserIdentifier = name.getOrElse( email.getOrElse( @@ -40,7 +40,7 @@ ) } - val empty = GerritUserIdentifiers(Map.empty[GerriAccountId, GerritUserIdentifier]) + val empty = GerritUserIdentifiers(Map.empty[GerritAccountId, GerritUserIdentifier]) private val gerritAccountSerializer = FieldSerializer[GerritAccount]( deserializer=renameFrom(name="_account_id",newName="accountId")
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/package.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/package.scala index 4cc45a8..0624eae 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/package.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/broadcast/package.scala
@@ -15,6 +15,6 @@ package com.gerritforge.analytics.auditlog package object broadcast { - type GerriAccountId = Int + type GerritAccountId = Int type GerritUserIdentifier = String }
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/job/Main.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/job/Main.scala index adda7f9..8690c58 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/job/Main.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/job/Main.scala
@@ -14,7 +14,7 @@ package com.gerritforge.analytics.auditlog.job -import com.gerritforge.analytics.auditlog.broadcast.GerritUserIdentifiers +import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUserInfo, GerritUserIdentifiers} import com.gerritforge.analytics.auditlog.model.ElasticSearchFields._ import com.gerritforge.analytics.auditlog.model._ import com.gerritforge.analytics.auditlog.range.TimeRange @@ -40,14 +40,21 @@ sys.exit(1) } + val triedAdditionalUserInfo = AdditionalUserInfo.loadAdditionalUserInfo(config) + if (triedAdditionalUserInfo.isFailure) { + logger.error("Error loading additional user information", triedAdditionalUserInfo.failed.get) + sys.exit(1) + } + spark .getEventsFromPath(config.eventsPath.get) - .transformEvents(tryUserIdentifiers.get, config.eventsTimeAggregation.get, TimeRange(config.since, config.until)) + .transformEvents(tryUserIdentifiers.get, triedAdditionalUserInfo.get,config.eventsTimeAggregation.get, TimeRange(config.since, config.until)) .saveToEs(s"${config.elasticSearchIndex.get}/$DOCUMENT_TYPE") case None => logger.error("Could not parse command line arguments") sys.exit(1) } + }
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AuditLogETLConfig.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AuditLogETLConfig.scala index 30e4c93..b95d81a 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AuditLogETLConfig.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AuditLogETLConfig.scala
@@ -25,5 +25,6 @@ since: Option[LocalDate] = None, until: Option[LocalDate] = None, eventsPath: Option[String] = None, - eventsTimeAggregation: Option[String] = Some("hour") + eventsTimeAggregation: Option[String] = Some("hour"), + additionalUserInfoPath: Option[String] = None )
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/CommandLineArguments.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/CommandLineArguments.scala index 2ff6854..21b6a84 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/CommandLineArguments.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/CommandLineArguments.scala
@@ -46,6 +46,10 @@ c.copy(eventsPath = Some(input)) } text "path to a directory (or a file) containing auditLogs events. Supports also '.gz' files. (Required)" + opt[String]('a', "additionalUserInfoPath") required () action { (input, c) => + c.copy(additionalUserInfoPath = Some(input)) + } text "path to a CSV file containing additional user information (Optional)\n\t\t\t\tCSV must have an header. Example:\n\t\t\t\tid,type\n\t\t\t\t123456,'bot'\n\t\t\t\t678876,'human'" + opt[String]('a', "eventsTimeAggregation") optional () action { (input, c) => c.copy(eventsTimeAggregation = Some(input)) } text "Events of the same type, produced by the same user will be aggregated with this time granularity: " +
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/ElasticSearchFields.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/ElasticSearchFields.scala index 41d33ec..1980385 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/ElasticSearchFields.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/ElasticSearchFields.scala
@@ -22,11 +22,13 @@ val AUDIT_TYPE_FIELD = "audit_type" val ACCESS_PATH_FIELD = "access_path" val RESULT_FIELD = "result" + val USER_TYPE_FIELD = "user_type" val FACETING_FIELDS = List( TIME_BUCKET_FIELD, AUDIT_TYPE_FIELD, USER_IDENTIFIER_FIELD, + USER_TYPE_FIELD, ACCESS_PATH_FIELD, COMMAND_FIELD, COMMAND_ARGS_FIELD,
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/AuditLogsTransformer.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/AuditLogsTransformer.scala index 1b06829..d9138bc 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/AuditLogsTransformer.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/AuditLogsTransformer.scala
@@ -14,7 +14,7 @@ package com.gerritforge.analytics.auditlog.spark -import com.gerritforge.analytics.auditlog.broadcast.GerritUserIdentifiers +import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUserInfo, AdditionalUsersInfo, GerritUserIdentifiers} import com.gerritforge.analytics.auditlog.model.AuditEvent import com.gerritforge.analytics.auditlog.model.ElasticSearchFields._ import com.gerritforge.analytics.auditlog.range.TimeRange @@ -23,9 +23,10 @@ import org.apache.spark.rdd.RDD import org.apache.spark.sql.{DataFrame, SparkSession} -case class AuditLogsTransformer(gerritIdentifiers: GerritUserIdentifiers = GerritUserIdentifiers.empty)(implicit spark: SparkSession) { +case class AuditLogsTransformer(gerritIdentifiers: GerritUserIdentifiers = GerritUserIdentifiers.empty, additionalUsersInfo: AdditionalUsersInfo = AdditionalUsersInfo.empty)(implicit spark: SparkSession) { private val broadcastUserIdentifiers = spark.sparkContext.broadcast(gerritIdentifiers) + private val broadcastAdditionalUsersInfo = spark.sparkContext.broadcast(additionalUsersInfo) def transform(auditEventsRDD: RDD[AuditEvent], timeAggregation: String, timeRange: TimeRange = TimeRange.always): DataFrame = auditEventsRDD @@ -35,5 +36,6 @@ .hydrateWithUserIdentifierColumn(USER_IDENTIFIER_FIELD, broadcastUserIdentifiers.value) .withTimeBucketColum(TIME_BUCKET_FIELD, timeAggregation) .withCommandColumns(COMMAND_FIELD, COMMAND_ARGS_FIELD) + .withUserTypeColumn(USER_TYPE_FIELD, broadcastAdditionalUsersInfo.value) .aggregateNumEventsColumn(NUM_EVENTS_FIELD, FACETING_FIELDS) -} +} \ No newline at end of file
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/dataframe/ops/DataFrameOps.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/dataframe/ops/DataFrameOps.scala index f1ddd9b..80f2e24 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/dataframe/ops/DataFrameOps.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/dataframe/ops/DataFrameOps.scala
@@ -14,7 +14,7 @@ package com.gerritforge.analytics.auditlog.spark.dataframe.ops -import com.gerritforge.analytics.auditlog.broadcast.GerritUserIdentifiers +import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUsersInfo, GerritUserIdentifiers} import com.gerritforge.analytics.auditlog.spark.sql.udf.SparkExtractors.{extractCommandArgumentsUDF, extractCommandUDF} import org.apache.spark.sql.{Column, DataFrame} import org.apache.spark.sql.expressions.UserDefinedFunction @@ -53,6 +53,12 @@ .withColumn(commandArgsCol, extractCommandArgumentsUDF(col("what"), col("access_path"))) } + def withUserTypeColumn(commandCol: String, additionalUsersInfo: AdditionalUsersInfo): DataFrame = { + def extractUserType: UserDefinedFunction = udf((who: Int) => additionalUsersInfo.getUserType(who)) + + dataFrame.withColumn(commandCol, ifExistThenGetOrNull("who", extractUserType(col("who")))) + } + def aggregateNumEventsColumn(numEventsCol: String, cols: List[String]): DataFrame = { dataFrame.groupBy(cols.map(c => col(c)): _*) .agg(count("*")
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/rdd/ops/SparkRDDOps.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/rdd/ops/SparkRDDOps.scala index 4602aee..22efd88 100644 --- a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/rdd/ops/SparkRDDOps.scala +++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/spark/rdd/ops/SparkRDDOps.scala
@@ -14,7 +14,7 @@ package com.gerritforge.analytics.auditlog.spark.rdd.ops -import com.gerritforge.analytics.auditlog.broadcast.GerritUserIdentifiers +import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUsersInfo, GerritUserIdentifiers} import com.gerritforge.analytics.auditlog.model.AuditEvent import com.gerritforge.analytics.auditlog.range.TimeRange import com.gerritforge.analytics.auditlog.spark.AuditLogsTransformer @@ -29,10 +29,12 @@ def toJsonString: RDD[String] = rdd.map(_.toJsonString) - def transformEvents(gerritUserIdentifiers: GerritUserIdentifiers, timeAggregation: String, timeRange: TimeRange) - (implicit spark: SparkSession): DataFrame = - AuditLogsTransformer(gerritUserIdentifiers) + def transformEvents(gerritUserIdentifiers: GerritUserIdentifiers, additionalUsersInfo: AdditionalUsersInfo, timeAggregation: String, timeRange: TimeRange) + (implicit spark: SparkSession): DataFrame = { + + AuditLogsTransformer(gerritUserIdentifiers, additionalUsersInfo) .transform(rdd, timeAggregation, timeRange) + } } implicit class PimpedStringRDD(rdd: RDD[String]) { @@ -41,5 +43,4 @@ spark.sqlContext.read.json(rdd.toDS()) } } - }
diff --git a/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala b/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala index aade9f9..b0a09a2 100644 --- a/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala +++ b/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala
@@ -16,7 +16,7 @@ import java.sql import com.gerritforge.analytics.SparkTestSupport -import com.gerritforge.analytics.auditlog.broadcast.GerritUserIdentifiers +import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUserInfo, AdditionalUsersInfo, GerritUserIdentifiers} import com.gerritforge.analytics.auditlog.model.{ElasticSearchFields, HttpAuditEvent, SshAuditEvent} import com.gerritforge.analytics.auditlog.spark.AuditLogsTransformer import com.gerritforge.analytics.support.ops.CommonTimeOperations._ @@ -38,6 +38,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), HttpAuditEvent.auditType, null, // no user identifier + null, // no user type anonymousHttpAuditEvent.accessPath.get, GIT_UPLOAD_PACK, anonymousHttpAuditEvent.what, @@ -58,6 +59,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), HttpAuditEvent.auditType, s"${authenticatedHttpAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, authenticatedHttpAuditEvent.accessPath.get, GIT_UPLOAD_PACK, authenticatedHttpAuditEvent.what, @@ -81,6 +83,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), HttpAuditEvent.auditType, gerritUserIdentifier, + AdditionalUserInfo.DEFAULT_USER_TYPE, authenticatedHttpAuditEvent.accessPath.get, GIT_UPLOAD_PACK, authenticatedHttpAuditEvent.what, @@ -101,6 +104,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), SshAuditEvent.auditType, s"${sshAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, sshAuditEvent.accessPath.get, SSH_GERRIT_COMMAND, SSH_GERRIT_COMMAND_ARGUMENTS, @@ -121,6 +125,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), SshAuditEvent.auditType, s"${sshAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, sshAuditEvent.accessPath.get, SSH_GERRIT_COMMAND, SSH_GERRIT_COMMAND_ARGUMENTS, @@ -143,6 +148,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), SshAuditEvent.auditType, s"${sshAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, sshAuditEvent.accessPath.get, SSH_GERRIT_COMMAND, SSH_GERRIT_COMMAND_ARGUMENTS, @@ -153,6 +159,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart), SshAuditEvent.auditType, s"${user2Id.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, sshAuditEvent.accessPath.get, SSH_GERRIT_COMMAND, SSH_GERRIT_COMMAND_ARGUMENTS, @@ -176,6 +183,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(events.head.timeAtStart), SshAuditEvent.auditType, s"${sshAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, sshAuditEvent.accessPath.get, SSH_GERRIT_COMMAND, SSH_GERRIT_COMMAND_ARGUMENTS, @@ -186,6 +194,7 @@ AuditLogsTransformerSpec.epochMillisToNearestHour(events.last.timeAtStart), HttpAuditEvent.auditType, s"${authenticatedHttpAuditEvent.who.get}", + AdditionalUserInfo.DEFAULT_USER_TYPE, authenticatedHttpAuditEvent.accessPath.get, GIT_UPLOAD_PACK, authenticatedHttpAuditEvent.what, @@ -194,6 +203,40 @@ ) ) } + + it should "process user type" in { + val events = Seq(authenticatedHttpAuditEvent) + + val userType = "nonDefaultUserType" + val additionalUserInfo = AdditionalUserInfo(authenticatedHttpAuditEvent.who.get, userType) + + val dataFrame = AuditLogsTransformer(additionalUsersInfo = AdditionalUsersInfo(Map(authenticatedHttpAuditEvent.who.get -> additionalUserInfo))).transform( + auditEventsRDD = sc.parallelize(events), + timeAggregation = "hour" + ) + dataFrame.collect.length shouldBe 1 + dataFrame.collect.head.getAs[String](ElasticSearchFields.USER_TYPE_FIELD) shouldBe userType + } + + it should "process user type when gerrit account could be identified" in { + val events = Seq(authenticatedHttpAuditEvent) + val gerritUserIdentifier = "Antonio Barone" + + val userType = "nonDefaultUserType" + val additionalUserInfo = AdditionalUserInfo(authenticatedHttpAuditEvent.who.get, userType) + + val dataFrame = + AuditLogsTransformer( + gerritIdentifiers = GerritUserIdentifiers(Map(authenticatedHttpAuditEvent.who.get -> gerritUserIdentifier)), + additionalUsersInfo = AdditionalUsersInfo(Map(authenticatedHttpAuditEvent.who.get -> additionalUserInfo)) + ).transform( + auditEventsRDD = sc.parallelize(events), + timeAggregation = "hour" + ) + + dataFrame.collect.length shouldBe 1 + dataFrame.collect.head.getAs[String](ElasticSearchFields.USER_TYPE_FIELD) shouldBe userType + } } object AuditLogsTransformerSpec {