Deal with DataSet[AggregatedAuditEvent] in Auditlog
Previous this change, the auditlog ETL job was exposing transformed data
as dataframe, this changes modifies that interface to deal with
Dataset[AggregatedAuditEvent] instead.
An AggregatedAuditEvent represents the result of spark transformations
and aggregations before being indexed in elasticsearch.
Change-Id: Ic3e8b8aa1b9e4294b3500de1d98e66623188f5d1
diff --git a/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AggregatedAuditEvent.scala b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AggregatedAuditEvent.scala
new file mode 100644
index 0000000..2fd3331
--- /dev/null
+++ b/auditlog/src/main/scala/com/gerritforge/analytics/auditlog/model/AggregatedAuditEvent.scala
@@ -0,0 +1,29 @@
+// Copyright (C) 2019 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.auditlog.model
+import java.sql.Timestamp
+
+case class AggregatedAuditEvent(
+ events_time_bucket: Timestamp,
+ audit_type: String,
+ user_identifier: Option[String],
+ user_type: Option[String],
+ access_path: Option[String],
+ command: String,
+ command_arguments: String,
+ project: Option[String],
+ result: String,
+ num_events: Long
+)
\ No newline at end of file
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 3fef687..9710149 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
@@ -15,13 +15,13 @@
package com.gerritforge.analytics.auditlog.spark
import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUsersInfo, GerritProjects, GerritUserIdentifiers}
-import com.gerritforge.analytics.auditlog.model.AuditEvent
+import com.gerritforge.analytics.auditlog.model.{AggregatedAuditEvent, AuditEvent}
import com.gerritforge.analytics.auditlog.model.ElasticSearchFields._
import com.gerritforge.analytics.auditlog.range.TimeRange
import com.gerritforge.analytics.auditlog.spark.dataframe.ops.DataFrameOps._
import com.gerritforge.analytics.auditlog.spark.rdd.ops.SparkRDDOps._
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.{Dataset, SparkSession}
case class AuditLogsTransformer(
gerritIdentifiers: GerritUserIdentifiers = GerritUserIdentifiers.empty,
@@ -33,7 +33,8 @@
private val broadcastAdditionalUsersInfo = spark.sparkContext.broadcast(additionalUsersInfo)
private val broadcastGerritProjects = spark.sparkContext.broadcast(gerritProjects)
- def transform(auditEventsRDD: RDD[AuditEvent], timeAggregation: String, timeRange: TimeRange = TimeRange.always): DataFrame =
+ def transform(auditEventsRDD: RDD[AuditEvent], timeAggregation: String, timeRange: TimeRange = TimeRange.always): Dataset[AggregatedAuditEvent] = {
+ import spark.implicits._
auditEventsRDD
.filterWithinRange(TimeRange(timeRange.since, timeRange.until))
.toJsonString
@@ -44,4 +45,6 @@
.withUserTypeColumn(USER_TYPE_FIELD, broadcastAdditionalUsersInfo.value)
.withProjectColumn(PROJECT_FIELD, broadcastGerritProjects.value)
.aggregateNumEventsColumn(NUM_EVENTS_FIELD, FACETING_FIELDS)
-}
\ No newline at end of file
+ .as[AggregatedAuditEvent]
+ }
+}
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 bfc466e..3887aca 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
@@ -15,11 +15,11 @@
package com.gerritforge.analytics.auditlog.spark.rdd.ops
import com.gerritforge.analytics.auditlog.broadcast.{AdditionalUsersInfo, GerritProjects, GerritUserIdentifiers}
-import com.gerritforge.analytics.auditlog.model.AuditEvent
+import com.gerritforge.analytics.auditlog.model.{AggregatedAuditEvent, AuditEvent}
import com.gerritforge.analytics.auditlog.range.TimeRange
import com.gerritforge.analytics.auditlog.spark.AuditLogsTransformer
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
object SparkRDDOps {
@@ -35,7 +35,7 @@
gerritProjects: GerritProjects,
timeAggregation: String,
timeRange: TimeRange
- )(implicit spark: SparkSession): DataFrame = {
+ )(implicit spark: SparkSession): Dataset[AggregatedAuditEvent] = {
AuditLogsTransformer(gerritUserIdentifiers, additionalUsersInfo, gerritProjects)
.transform(rdd, timeAggregation, timeRange)
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 3b5fefc..79df0da 100644
--- a/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala
+++ b/auditlog/src/test/scala/com/gerritforge/analytics/auditlog/AuditLogsTransformerSpec.scala
@@ -17,10 +17,9 @@
import com.gerritforge.analytics.SparkTestSupport
import com.gerritforge.analytics.auditlog.broadcast._
-import com.gerritforge.analytics.auditlog.model.{ElasticSearchFields, HttpAuditEvent, SshAuditEvent}
+import com.gerritforge.analytics.auditlog.model.{AggregatedAuditEvent, ElasticSearchFields, HttpAuditEvent, SshAuditEvent}
import com.gerritforge.analytics.auditlog.spark.AuditLogsTransformer
import com.gerritforge.analytics.support.ops.CommonTimeOperations._
-import org.apache.spark.sql.Row
import org.scalatest.{FlatSpec, Matchers}
class AuditLogsTransformerSpec extends FlatSpec with Matchers with SparkTestSupport with TestFixtures {
@@ -29,44 +28,42 @@
it should "process an anonymous http audit entry" in {
val events = Seq(anonymousHttpAuditEvent)
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 1
- dataFrame.collect should contain only Row(
- AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
- HttpAuditEvent.auditType,
- null, // no user identifier
- null, // no user type
- anonymousHttpAuditEvent.accessPath.get,
- GIT_UPLOAD_PACK,
- anonymousHttpAuditEvent.what,
- null, // no project
- anonymousHttpAuditEvent.result,
- expectedAggregatedCount
+ aggregatedEventsDS.collect should contain only AggregatedAuditEvent(
+ AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
+ HttpAuditEvent.auditType,
+ None,
+ None,
+ anonymousHttpAuditEvent.accessPath,
+ GIT_UPLOAD_PACK,
+ anonymousHttpAuditEvent.what,
+ None,
+ anonymousHttpAuditEvent.result,
+ num_events = 1
)
}
it should "process an authenticated http audit entry where gerrit account couldn't be identified" in {
val events = Seq(authenticatedHttpAuditEvent)
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 1
- dataFrame.collect should contain only Row(
+ aggregatedEventsDS.collect should contain only AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
HttpAuditEvent.auditType,
- s"${authenticatedHttpAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- authenticatedHttpAuditEvent.accessPath.get,
+ authenticatedHttpAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ anonymousHttpAuditEvent.accessPath,
GIT_UPLOAD_PACK,
- authenticatedHttpAuditEvent.what,
- null, // no project
- authenticatedHttpAuditEvent.result,
- expectedAggregatedCount
+ anonymousHttpAuditEvent.what,
+ None,
+ anonymousHttpAuditEvent.result,
+ num_events = 1
)
}
@@ -74,68 +71,65 @@
val events = Seq(authenticatedHttpAuditEvent)
val gerritUserIdentifier = "Antonio Barone"
- val dataFrame =
+ val aggregatedEventsDS =
AuditLogsTransformer(GerritUserIdentifiers(Map(authenticatedHttpAuditEvent.who.get -> gerritUserIdentifier)))
.transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 1
- dataFrame.collect should contain only Row(
+ aggregatedEventsDS.collect should contain only AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
HttpAuditEvent.auditType,
- gerritUserIdentifier,
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- authenticatedHttpAuditEvent.accessPath.get,
+ Some(gerritUserIdentifier),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ anonymousHttpAuditEvent.accessPath,
GIT_UPLOAD_PACK,
authenticatedHttpAuditEvent.what,
- null, // no project
+ None,
authenticatedHttpAuditEvent.result,
- expectedAggregatedCount
+ num_events = 1
)
}
it should "process an SSH audit entry" in {
val events = Seq(sshAuditEvent)
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 1
- dataFrame.collect should contain only Row(
+ aggregatedEventsDS.collect should contain only AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
SshAuditEvent.auditType,
- s"${sshAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- sshAuditEvent.accessPath.get,
+ sshAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ sshAuditEvent.accessPath,
SSH_GERRIT_COMMAND,
SSH_GERRIT_COMMAND_ARGUMENTS,
- null, // no project
+ None,
sshAuditEvent.result,
- expectedAggregatedCount
+ num_events = 1
)
}
it should "group ssh events from the same user together, if they fall within the same time bucket (hour)" in {
val events = Seq(sshAuditEvent, sshAuditEvent.copy(timeAtStart = timeAtStart + 1000))
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 2
- dataFrame.collect should contain only Row(
+ aggregatedEventsDS.collect should contain only AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
SshAuditEvent.auditType,
- s"${sshAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- sshAuditEvent.accessPath.get,
+ sshAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ sshAuditEvent.accessPath,
SSH_GERRIT_COMMAND,
SSH_GERRIT_COMMAND_ARGUMENTS,
- null, // no project
+ None,
sshAuditEvent.result,
- expectedAggregatedCount
+ num_events = 2
)
}
@@ -143,35 +137,34 @@
val user2Id = sshAuditEvent.who.map(_ + 1)
val events = Seq(sshAuditEvent, sshAuditEvent.copy(who=user2Id))
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedAggregatedCount = 1
- dataFrame.collect should contain allOf (
- Row(
+ aggregatedEventsDS.collect should contain allOf (
+ AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
SshAuditEvent.auditType,
- s"${sshAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- sshAuditEvent.accessPath.get,
+ sshAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ sshAuditEvent.accessPath,
SSH_GERRIT_COMMAND,
SSH_GERRIT_COMMAND_ARGUMENTS,
- null, // no project
+ None,
sshAuditEvent.result,
- expectedAggregatedCount
+ num_events = 1
),
- Row(
+ AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(timeAtStart),
SshAuditEvent.auditType,
- s"${user2Id.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- sshAuditEvent.accessPath.get,
+ user2Id.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ sshAuditEvent.accessPath,
SSH_GERRIT_COMMAND,
SSH_GERRIT_COMMAND_ARGUMENTS,
- null, // no project
+ None,
sshAuditEvent.result,
- expectedAggregatedCount
+ num_events = 1
)
)
}
@@ -179,36 +172,34 @@
it should "group different event types separately, event if they fall within the same time bucket (hour)" in {
val events = Seq(sshAuditEvent, authenticatedHttpAuditEvent.copy(timeAtStart = sshAuditEvent.timeAtStart + 1000))
- val dataFrame = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
+ val aggregatedEventsDS = AuditLogsTransformer().transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
+ aggregatedEventsDS.columns should contain theSameElementsAs ElasticSearchFields.ALL_DOCUMENT_FIELDS
- val expectedSshAggregatedCount = 1
- val expectedHttpAggregatedCount = 1
- dataFrame.collect should contain allOf (
- Row(
+ aggregatedEventsDS.collect should contain allOf (
+ AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(events.head.timeAtStart),
SshAuditEvent.auditType,
- s"${sshAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- sshAuditEvent.accessPath.get,
+ sshAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ sshAuditEvent.accessPath,
SSH_GERRIT_COMMAND,
SSH_GERRIT_COMMAND_ARGUMENTS,
- null, // no project
+ None,
sshAuditEvent.result,
- expectedSshAggregatedCount
+ num_events = 1
),
- Row(
+ AggregatedAuditEvent(
AuditLogsTransformerSpec.epochMillisToNearestHour(events.last.timeAtStart),
HttpAuditEvent.auditType,
- s"${authenticatedHttpAuditEvent.who.get}",
- AdditionalUserInfo.DEFAULT_USER_TYPE,
- authenticatedHttpAuditEvent.accessPath.get,
+ authenticatedHttpAuditEvent.who.map(_.toString),
+ Some(AdditionalUserInfo.DEFAULT_USER_TYPE),
+ authenticatedHttpAuditEvent.accessPath,
GIT_UPLOAD_PACK,
authenticatedHttpAuditEvent.what,
- null, // no project
+ None,
authenticatedHttpAuditEvent.result,
- expectedHttpAggregatedCount
+ num_events = 1
)
)
}
@@ -219,12 +210,12 @@
val userType = "nonDefaultUserType"
val additionalUserInfo = AdditionalUserInfo(authenticatedHttpAuditEvent.who.get, userType)
- val dataFrame = AuditLogsTransformer(additionalUsersInfo = AdditionalUsersInfo(Map(authenticatedHttpAuditEvent.who.get -> additionalUserInfo))).transform(
+ val aggregatedEventsDS = 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
+ aggregatedEventsDS.collect.length shouldBe 1
+ aggregatedEventsDS.collect.head.user_type should contain(userType)
}
it should "process user type when gerrit account could be identified" in {
@@ -234,7 +225,7 @@
val userType = "nonDefaultUserType"
val additionalUserInfo = AdditionalUserInfo(authenticatedHttpAuditEvent.who.get, userType)
- val dataFrame =
+ val aggregatedEventsDS =
AuditLogsTransformer(
gerritIdentifiers = GerritUserIdentifiers(Map(authenticatedHttpAuditEvent.who.get -> gerritUserIdentifier)),
additionalUsersInfo = AdditionalUsersInfo(Map(authenticatedHttpAuditEvent.who.get -> additionalUserInfo))
@@ -243,28 +234,28 @@
timeAggregation = "hour"
)
- dataFrame.collect.length shouldBe 1
- dataFrame.collect.head.getAs[String](ElasticSearchFields.USER_TYPE_FIELD) shouldBe userType
+ aggregatedEventsDS.collect.length shouldBe 1
+ aggregatedEventsDS.collect.head.user_type should contain(userType)
}
it should "extract gerrit project from an http event" in {
val events = Seq(authenticatedHttpAuditEvent)
- val dataFrame = AuditLogsTransformer(gerritProjects = GerritProjects(Map(project -> GerritProject(project))))
+ val aggregatedEventsDS = AuditLogsTransformer(gerritProjects = GerritProjects(Map(project -> GerritProject(project))))
.transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.collect.length shouldBe 1
- dataFrame.collect.head.getAs[String](ElasticSearchFields.PROJECT_FIELD) shouldBe project
+ aggregatedEventsDS.collect.length shouldBe 1
+ aggregatedEventsDS.collect.head.project should contain(project)
}
it should "extract gerrit project from an ssh event" in {
val events = Seq(sshAuditEvent)
- val dataFrame = AuditLogsTransformer(gerritProjects = GerritProjects(Map(project -> GerritProject(project))))
+ val aggregatedEventsDS = AuditLogsTransformer(gerritProjects = GerritProjects(Map(project -> GerritProject(project))))
.transform(sc.parallelize(events), timeAggregation="hour")
- dataFrame.collect.length shouldBe 1
- dataFrame.collect.head.getAs[String](ElasticSearchFields.PROJECT_FIELD) shouldBe project
+ aggregatedEventsDS.collect.length shouldBe 1
+ aggregatedEventsDS.collect.head.project should contain(project)
}
}
