You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by ra...@apache.org on 2017/06/06 20:00:43 UTC
[1/8] incubator-spot git commit: flow_tuning
Repository: incubator-spot
Updated Branches:
refs/heads/master 01b1de758 -> c0f80c56e
flow_tuning
experimental branch for the tuning of netflow:
* shall we incorporate protocol information?
* shall we bin time by hour ?
* shall we bin bytes by exponential buckets (eg log of bytes)
* shall we bin packet coutns by exponential buckets (eg log of packet counts) ?
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/8c6a530d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/8c6a530d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/8c6a530d
Branch: refs/heads/master
Commit: 8c6a530d58fc4a32c1d5295af84653a6cd111d2b
Parents: 77d6081
Author: nlsegerl <na...@intel.com>
Authored: Thu Apr 27 14:00:13 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Thu Apr 27 14:00:13 2017 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/sql/WideUDFs.scala | 18 ++
.../spot/SuspiciousConnectsArgumentParser.scala | 28 ++-
.../apache/spot/netflow/FlowWordCreator.scala | 65 ++++--
.../spot/netflow/model/FlowScoreFunction.scala | 12 +-
.../model/FlowSuspiciousConnectsModel.scala | 134 +++++++----
.../spot/netflow/FlowWordCreatorTest.scala | 233 +++++++++++++++----
6 files changed, 378 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala b/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
index ba6e4af..28ecdc2 100644
--- a/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
+++ b/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
@@ -47,4 +47,22 @@ object WideUDFs {
:: Nil).getOrElse(Nil)
UserDefinedFunction(f, ScalaReflection.schemaFor(typeTag[RT]).dataType, inputTypes)
}
+ def udf[RT: TypeTag, A1: TypeTag, A2: TypeTag, A3: TypeTag, A4: TypeTag, A5: TypeTag, A6: TypeTag,
+ A7: TypeTag, A8: TypeTag, A9: TypeTag, A10: TypeTag, A11: TypeTag, A12: TypeTag](f: Function12[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, RT]):
+ UserDefinedFunction = {
+ val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A2]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A3]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A4]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A5]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A6]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A7]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A8]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A9]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A10]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A11]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A12]).dataType
+ :: Nil).getOrElse(Nil)
+ UserDefinedFunction(f, ScalaReflection.schemaFor(typeTag[RT]).dataType, inputTypes)
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
index 1966dec..bf9c710 100644
--- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
@@ -36,7 +36,14 @@ object SuspiciousConnectsArgumentParser {
ldaPRGSeed: Option[Long] = None,
ldaMaxiterations: Int = 20,
ldaAlpha: Double = 1.02,
- ldaBeta: Double = 1.001)
+ ldaBeta: Double = 1.001,
+ // flow tuning options
+ // these are for the flow tuning branch only!
+
+ flatBinTime: Boolean = false,
+ expBinIBytes: Boolean = false,
+ expBinIPkts: Boolean = false,
+ useProtocol: Boolean = false)
val parser: scopt.OptionParser[SuspiciousConnectsConfig] = new scopt.OptionParser[SuspiciousConnectsConfig]("LDA") {
@@ -99,5 +106,24 @@ object SuspiciousConnectsArgumentParser {
opt[Double]("ldabeta").optional().valueName("float64").
action((x, c) => c.copy(ldaBeta = x)).
text("topic concentration for lda, default 1.001")
+
+
+ // these are for the flow tuning branch only!
+ opt[Boolean]("hourbins").optional().
+ action((x, c) => c.copy(flatBinTime = true)).
+ text("use per-hour binning strategy for flow time of day")
+
+ opt[Boolean]("expbinbytes").optional().
+ action((x, c) => c.copy(expBinIBytes = true)).
+ text("use exponential binning strategy for flow byte counts")
+
+ opt[Boolean]("expbinpkts").optional().
+ action((x, c) => c.copy(expBinIPkts = true)).
+ text("use exponential binning strategy for flow packet counts")
+
+ opt[Boolean]("protocol").optional().
+ action((x, c) => c.copy(useProtocol = true)).
+ text("incorporate flow protocol into netflow word creation")
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
index e091416..4ff716e 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
@@ -42,7 +42,11 @@ case class FlowWords(srcWord: String, dstWord: String)
*/
class FlowWordCreator(timeCuts: Array[Double],
ibytCuts: Array[Double],
- ipktCuts: Array[Double]) extends Serializable {
+ ipktCuts: Array[Double],
+ useProtocol: Boolean = false,
+ hourBinTime: Boolean = false,
+ expBinBytes: Boolean = false,
+ expBinPackets: Boolean =false ) extends Serializable {
/**
@@ -57,9 +61,10 @@ class FlowWordCreator(timeCuts: Array[Double],
dstIP: String,
srcPort: Int,
dstPort: Int,
- ipkt: Long,
- ibyt: Long) =>
- flowWords(hour, minute, second, srcPort, dstPort, ipkt, ibyt).srcWord)
+ protocol: String,
+ ibyt: Long,
+ ipkt: Long) =>
+ flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyt, ipkt).srcWord)
/**
@@ -74,9 +79,10 @@ class FlowWordCreator(timeCuts: Array[Double],
dstIP: String,
srcPort: Int,
dstPort: Int,
- ipkt: Long,
- ibyt: Long) =>
- flowWords(hour, minute, second, srcPort, dstPort, ipkt, ibyt).dstWord)
+ protocol: String,
+ ibyt: Long,
+ ipkt: Long) =>
+ flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyt, ipkt).dstWord)
/**
@@ -91,53 +97,76 @@ class FlowWordCreator(timeCuts: Array[Double],
* @param ibyt
* @return [[FlowWords]] containing source and destination words.
*/
- def flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, ipkt: Long, ibyt: Long): FlowWords = {
+ def flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol: String, ibyt: Long, ipkt: Long): FlowWords = {
Try {
val timeOfDay: Double = hour.toDouble + minute.toDouble / 60 + second.toDouble / 3600
- val timeBin = Quantiles.bin(timeOfDay, timeCuts)
- val ibytBin = Quantiles.bin(ibyt, ibytCuts)
- val ipktBin = Quantiles.bin(ipkt, ipktCuts)
+ val timeBin = if (hourBinTime == true) {
+ hour
+ } else {
+ Quantiles.bin(timeOfDay, timeCuts)
+ }
+
+ val lnOf2 = scala.math.log(2) // natural log of 2
+ val ibytBin : Long = if (expBinBytes) {
+ scala.math.ceil(scala.math.log(ibyt) / lnOf2).toLong // 0 values should never ever happen
+ } else {
+ Quantiles.bin(ibyt, ibytCuts)
+
+ }
+ val ipktBin : Long = if (expBinPackets) {
+ scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
+ } else {
+ Quantiles.bin(ipkt, ibytCuts)
+ }
val LowToLowPortEncoding = 111111
val HighToHighPortEncoding = 333333
+ val proto = if (useProtocol == true) {
+ protocol
+ } else {
+ ""
+ }
+
if (dstPort == 0 && srcPort == 0) {
- val baseWord = Array("0", timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array("0", proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
} else if (dstPort == 0 && srcPort > 0) {
- val baseWord = Array(srcPort.toString(), timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(srcPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = "-1_" + baseWord, dstWord = baseWord)
} else if (srcPort == 0 && dstPort > 0) {
- val baseWord = Array(dstPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else if (srcPort <= 1024 && dstPort <= 1024) {
- val baseWord = Array(LowToLowPortEncoding, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(LowToLowPortEncoding, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
} else if (srcPort <= 1024 && dstPort > 1024) {
- val baseWord = Array(srcPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(srcPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = "-1_" + baseWord, dstWord = baseWord)
} else if (srcPort > 1024 && dstPort <= 1024) {
- val baseWord = Array(dstPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else {
// this is the srcPort > 1024 && dstPort > 1024 case
- val baseWord = Array(HighToHighPortEncoding, timeBin, ibytBin, ipktBin).mkString("_")
+
+
+ val baseWord = Array(HighToHighPortEncoding, proto, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
index 2ee29a8..63152e5 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
@@ -34,14 +34,18 @@ import org.apache.spot.utilities.data.validation.InvalidDataHandler
*/
-class FlowScoreFunction(timeCuts: Array[Double],
+class FlowScoreFunction(useProtocol: Boolean,
+ hourBinTime : Boolean,
+ expBinBytes : Boolean,
+ expBinPackets: Boolean,
+ timeCuts: Array[Double],
ibytCuts: Array[Double],
ipktCuts: Array[Double],
topicCount: Int,
wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable {
- val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts)
+ val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, useProtocol, hourBinTime, expBinBytes, expBinPackets)
/**
* Estimate the probability of a netflow connection as distributed from the source IP and from the destination IP
@@ -67,14 +71,14 @@ class FlowScoreFunction(timeCuts: Array[Double],
dstIP: String,
srcPort: Int,
dstPort: Int,
+ protocol: String,
ipkt: Long,
ibyt: Long,
srcTopicMix: Seq[Double],
dstTopicMix: Seq[Double]): Double = {
- val FlowWords(srcWord, dstWord) = flowWordCreator.flowWords(hour: Int, minute: Int, second: Int,
- srcPort: Int, dstPort: Int, ipkt: Long, ibyt: Long)
+ val FlowWords(srcWord, dstWord) = flowWordCreator.flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol, ibyt: Long, ipkt: Long)
val zeroProb = Array.fill(topicCount) {
0.0
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index b5fdda5..95dca42 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -61,6 +61,10 @@ import scala.util.{Failure, Success, Try}
class FlowSuspiciousConnectsModel(topicCount: Int,
ipToTopicMix: DataFrame,
wordToPerTopicProb: Map[String, Array[Double]],
+ useProtocol: Boolean,
+ hourBinTime: Boolean,
+ expBinBytes: Boolean,
+ expBinPackets: Boolean,
timeCuts: Array[Double],
ibytCuts: Array[Double],
ipktCuts: Array[Double]) {
@@ -87,7 +91,11 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
recordsWithIPTopicMixes.selectExpr(schema: _*).withColumnRenamed(TopicProbabilityMix, DstIpTopicMix)
}
- val scoreFunction = new FlowScoreFunction(timeCuts,
+ val scoreFunction = new FlowScoreFunction(useProtocol,
+ hourBinTime,
+ expBinBytes,
+ expBinPackets,
+ timeCuts,
ibytCuts,
ipktCuts,
topicCount,
@@ -101,6 +109,7 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
dstIP: String,
srcPort: Int,
dstPort: Int,
+ protocol: String,
ipkt: Long,
ibyt: Long,
srcIpTopicMix: Seq[Double],
@@ -112,6 +121,7 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
dstIP,
srcPort,
dstPort,
+ protocol,
ipkt,
ibyt,
srcIpTopicMix,
@@ -139,8 +149,9 @@ object FlowSuspiciousConnectsModel {
DestinationIPField,
SourcePortField,
DestinationPortField,
- IpktField,
- IbytField))
+ ProtocolField,
+ IbytField,
+ IpktField))
val ModelColumns = ModelSchema.fieldNames.toList.map(col)
@@ -163,55 +174,80 @@ object FlowSuspiciousConnectsModel {
// create quantile cut-offs
- val timeCuts = Quantiles.computeDeciles(totalRecords
- .select(Hour, Minute, Second)
- .rdd
- .flatMap({ case Row(hours: Int, minutes: Int, seconds: Int) =>
- Try {
- (3600 * hours + 60 * minutes + seconds).toDouble
- } match {
- case Failure(_) => Seq()
- case Success(time) => Seq(time)
- }
- }))
-
- logger.info(timeCuts.mkString(","))
-
- logger.info("calculating byte cuts ...")
-
- val ibytCuts = Quantiles.computeDeciles(totalRecords
- .select(Ibyt)
- .rdd
- .flatMap({ case Row(ibyt: Long) =>
- Try {
- ibyt.toDouble
- } match {
- case Failure(_) => Seq()
- case Success(ibyt) => Seq(ibyt)
- }
- }))
-
- logger.info(ibytCuts.mkString(","))
+ val timeCuts = if (config.flatBinTime == false) {
+ logger.info("Using decile based time binning")
+ val timeCuts = Quantiles.computeDeciles(totalRecords
+ .select(Hour, Minute, Second)
+ .rdd
+ .flatMap({ case Row(hours: Int, minutes: Int, seconds: Int) =>
+ Try {
+ (3600 * hours + 60 * minutes + seconds).toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(time) => Seq(time)
+ }
+ }))
+
+ logger.info(timeCuts.mkString(","))
+ timeCuts
+ } else {
+ logger.info("Using hour-based time binning")
+ null
+ }
+
+
+
- logger.info("calculating pkt cuts")
- val ipktCuts = Quantiles.computeQuintiles(totalRecords
- .select(Ipkt)
- .rdd
- .flatMap({ case Row(ipkt: Long) =>
- Try {
- ipkt.toDouble
- } match {
- case Failure(_) => Seq()
- case Success(ipkt) => Seq(ipkt)
- }
- }))
+ val ibytCuts = if (config.expBinIBytes == false) {
+ logger.info("Using decile based ibytes binning")
+ val ibytCuts = Quantiles.computeDeciles(totalRecords
+ .select(Ibyt)
+ .rdd
+ .flatMap({ case Row(ibyt: Long) =>
+ Try {
+ ibyt.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(ibyt) => Seq(ibyt)
+ }
+ }))
+
+ logger.info(ibytCuts.mkString(","))
+ ibytCuts
+ } else {
+ logger.info("using exponential binning strategy for flow ibytes")
+ null
+ }
+
- logger.info(ipktCuts.mkString(","))
- // simplify DNS log entries into "words"
+ logger.info("calculating pkt cuts")
+
+ val ipktCuts = if (config.expBinIPkts == false) {
+ logger.info("Using quintile based ipkt count binning")
+ val ipktCuts =
+ Quantiles.computeQuintiles(totalRecords
+ .select(Ipkt)
+ .rdd
+ .flatMap({ case Row(ipkt: Long) =>
+ Try {
+ ipkt.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(ipkt) => Seq(ipkt)
+ }
+ }))
+
+ logger.info(ipktCuts.mkString(","))
+ ipktCuts
+ } else {
+ logger.info("Using exponential binning for ipkts")
+ null
+ }
+ // simplify dflow log entries into "words"
- val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts)
+ val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, config.useProtocol, config.flatBinTime, config.expBinIBytes, config.expBinIPkts)
val dataWithWords = totalRecords.withColumn(SourceWord, flowWordCreator.srcWordUDF(ModelColumns: _*))
.withColumn(DestinationWord, flowWordCreator.dstWordUDF(ModelColumns: _*))
@@ -258,6 +294,10 @@ object FlowSuspiciousConnectsModel {
new FlowSuspiciousConnectsModel(topicCount,
ipToTopicMix,
wordToPerTopicProb,
+ config.useProtocol,
+ config.flatBinTime,
+ config.expBinIBytes,
+ config.expBinIPkts,
timeCuts,
ibytCuts,
ipktCuts)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/8c6a530d/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
index 9e7f6a9..a3d8a0a 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
@@ -29,6 +29,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val minute = 59
val second = 32
+ val protocol = "UDP"
val ibyts = 222L
val ipkts = 3L
@@ -50,11 +51,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 23
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_23_5_2_0"
- srcWord shouldBe "23_5_2_0"
+ dstWord shouldBe "-1_23__5_2_0"
+ srcWord shouldBe "23__5_2_0"
}
@@ -65,10 +66,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2132
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "23_5_2_0"
- srcWord shouldBe "-1_23_5_2_0"
+ dstWord shouldBe "23__5_2_0"
+ srcWord shouldBe "-1_23__5_2_0"
}
// 3. Test when sip is less than dip and sip is not 0 and dport and sport are > 1024 +
@@ -77,10 +78,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 9874
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "333333_5_2_0"
- srcWord shouldBe "333333_5_2_0"
+ dstWord shouldBe "333333__5_2_0"
+ srcWord shouldBe "333333__5_2_0"
}
// 4. Test when sip is less than dip and sip is not 0 and dport is 0 but sport is not +
@@ -89,11 +90,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 0
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80_5_2_0"
- srcWord shouldBe "-1_80_5_2_0"
+ dstWord shouldBe "80__5_2_0"
+ srcWord shouldBe "-1_80__5_2_0"
}
// 5. Test when sip is less than dip and sip is not 0 and sport is 0 but dport is not +
@@ -104,11 +105,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_43_5_2_0"
- srcWord shouldBe "43_5_2_0"
+ dstWord shouldBe "-1_43__5_2_0"
+ srcWord shouldBe "43__5_2_0"
}
// 6. Test when sip is less than dip and sip is not 0 and sport and dport are less or equal than 1024 +
@@ -118,10 +119,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "111111_5_2_0"
- srcWord shouldBe "111111_5_2_0"
+ dstWord shouldBe "111111__5_2_0"
+ srcWord shouldBe "111111__5_2_0"
}
// 7. Test when sip is less than dip and sip is not 0 and sport and dport are 0+
@@ -131,10 +132,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "0_5_2_0"
- srcWord shouldBe "0_5_2_0"
+ dstWord shouldBe "0__5_2_0"
+ srcWord shouldBe "0__5_2_0"
}
// 8. Test when sip is not less than dip and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0+
@@ -144,10 +145,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_43_5_2_0"
- srcWord shouldBe "43_5_2_0"
+ dstWord shouldBe "-1_43__5_2_0"
+ srcWord shouldBe "43__5_2_0"
}
@@ -157,10 +158,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2435
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80_5_2_0"
- srcWord shouldBe "-1_80_5_2_0"
+ dstWord shouldBe "80__5_2_0"
+ srcWord shouldBe "-1_80__5_2_0"
}
@@ -171,10 +172,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "333333_5_2_0"
- srcWord shouldBe "333333_5_2_0"
+ dstWord shouldBe "333333__5_2_0"
+ srcWord shouldBe "333333__5_2_0"
}
// 11. Test when sip is not less than dip and dport is 0 but sport is not +
@@ -184,10 +185,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80_5_2_0"
- srcWord shouldBe "-1_80_5_2_0"
+ dstWord shouldBe "80__5_2_0"
+ srcWord shouldBe "-1_80__5_2_0"
}
// 12. Test when sip is not less than dip and sport is 0 but dport is not +
@@ -197,10 +198,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_2435_5_2_0"
- srcWord shouldBe "2435_5_2_0"
+ dstWord shouldBe "-1_2435__5_2_0"
+ srcWord shouldBe "2435__5_2_0"
}
// 13. Test when sip is not less than dip and sport and dport are less or equal than 1024
@@ -210,10 +211,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "111111_5_2_0"
- srcWord shouldBe "111111_5_2_0"
+ dstWord shouldBe "111111__5_2_0"
+ srcWord shouldBe "111111__5_2_0"
}
// 14. Test when sip is not less than dip and sport and dport are 0
@@ -223,9 +224,157 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, ipkts, ibyts)
+ flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "0_5_2_0"
- srcWord shouldBe "0_5_2_0"
+ dstWord shouldBe "0__5_2_0"
+ srcWord shouldBe "0__5_2_0"
+ }
+
+ // 15. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 and including protocol" in {
+ val srcPort = 2132
+ val dstPort = 23
+
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+
+ dstWord shouldBe "-1_23_TCP_5_2_0"
+ srcWord shouldBe "23_TCP_5_2_0"
+
+ }
+
+
+ // 16. Test when sip is not less than dip and sport and dport are 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) and including protocol" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0_TCP_5_2_0"
+ srcWord shouldBe "0_TCP_5_2_0"
+ }
+
+
+ // 17. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with protocol + modified bins" in {
+ val srcPort = 2132
+ val dstPort = 23
+
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true, true, true, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+
+ dstWord shouldBe "-1_23_UDP_12_8_2"
+ srcWord shouldBe "23_UDP_12_8_2"
+
+ }
+
+
+ // 18. Test when sip is not less than dip and sport and dport are 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with protocol + modified bins" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true, true, true, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0_TCP_12_8_2"
+ srcWord shouldBe "0_TCP_12_8_2"
+ }
+
+
+ // 19. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified time bins" in {
+ val srcPort = 2132
+ val dstPort = 23
+
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, true, false, false)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+
+ dstWord shouldBe "-1_23__12_2_0"
+ srcWord shouldBe "23__12_2_0"
+
+ }
+
+
+ // 20. Test when sip is not less than dip and sport and dport are 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified time bins" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, true, false, false)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0__12_2_0"
+ srcWord shouldBe "0__12_2_0"
+ }
+
+ // 21. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified byte bins" in {
+ val srcPort = 2132
+ val dstPort = 23
+
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, true, false)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+
+ dstWord shouldBe "-1_23__5_8_0"
+ srcWord shouldBe "23__5_8_0"
+
+ }
+
+
+ // 22. Test when sip is not less than dip and sport and dport are 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified byte bins" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, true,false)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0__5_8_0"
+ srcWord shouldBe "0__5_8_0"
+ }
+
+ // 23. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified packet bins" in {
+ val srcPort = 2132
+ val dstPort = 23
+
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, false, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+
+ dstWord shouldBe "-1_23__5_2_2"
+ srcWord shouldBe "23__5_2_2"
+
+ }
+
+
+ // 24. Test when sip is not less than dip and sport and dport are 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified packet bins" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, false, true)
+ val FlowWords(srcWord, dstWord) =
+ flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0__5_2_2"
+ srcWord shouldBe "0__5_2_2"
}
}
[6/8] incubator-spot git commit: Merge branch 'master' of
github.com:apache/incubator-spot into flow_tuning
Posted by ra...@apache.org.
Merge branch 'master' of github.com:apache/incubator-spot into flow_tuning
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/f18bd551
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/f18bd551
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/f18bd551
Branch: refs/heads/master
Commit: f18bd551534c02b4405a15f5c208b93a2c3ffd2d
Parents: e093531 01b1de7
Author: nlsegerl <na...@intel.com>
Authored: Fri Jun 2 12:32:57 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Fri Jun 2 12:32:57 2017 -0700
----------------------------------------------------------------------
.../proxy/ProxySuspiciousConnectsModel.scala | 74 +++-------
.../apache/spot/proxy/ProxyWordCreation.scala | 34 ++---
.../org/apache/spot/utilities/MathUtils.scala | 31 ++++
.../apache/spot/utilities/TimeUtilities.scala | 21 ++-
.../scala/org/apache/spot/QuantilesTest.scala | 2 +-
.../spot/proxy/ProxyWordCreationTest.scala | 148 +++++++++++++++++++
.../apache/spot/utilities/MathUtilsTest.scala | 40 +++++
.../spot/utilities/TimeUtilitiesTest.scala | 58 ++++++++
spot-oa/ui/dns/storyboard.html | 1 +
spot-oa/ui/dns/suspicious.html | 5 -
spot-oa/ui/flow/storyboard.html | 1 +
.../ui/js/components/DendrogramMixin.react.js | 35 ++++-
12 files changed, 365 insertions(+), 85 deletions(-)
----------------------------------------------------------------------
[2/8] incubator-spot git commit: flow_tuning
Posted by ra...@apache.org.
flow_tuning
some changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/f5f57851
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/f5f57851
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/f5f57851
Branch: refs/heads/master
Commit: f5f578518b3a505afc362c6a2940a68e51f1fe42
Parents: 8c6a530
Author: nlsegerl <na...@intel.com>
Authored: Tue May 16 13:28:16 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Tue May 16 13:28:16 2017 -0700
----------------------------------------------------------------------
.../spot/SuspiciousConnectsArgumentParser.scala | 2 +-
.../FlowSuspiciousConnectsAnalysis.scala | 40 ++++++--------------
spot-ml/src/scripts/TestingFlow.scala | 34 +++++++++++++++++
.../FlowSuspiciousConnectsAnalysisTest.scala | 19 ++++++----
4 files changed, 59 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/f5f57851/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
index bf9c710..e42a610 100644
--- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
@@ -26,7 +26,7 @@ object SuspiciousConnectsArgumentParser {
case class SuspiciousConnectsConfig(analysis: String = "",
inputPath: String = "",
feedbackFile: String = "",
- duplicationFactor: Int = 1,
+ duplicationFactor: Int = 1000,
topicCount: Int = 20,
userDomain: String = "",
hdfsScoredConnect: String = "",
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/f5f57851/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index b9421ae..4e97be1 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -37,13 +37,20 @@ object FlowSuspiciousConnectsAnalysis {
def run(config: SuspiciousConnectsConfig, sparkContext: SparkContext, sqlContext: SQLContext, logger: Logger,
- inputFlowRecords: DataFrame) = {
+ inputFlowRecords: DataFrame) : DataFrame = {
logger.info("Starting flow suspicious connects analysis.")
- val cleanFlowRecords = filterAndSelectCleanFlowRecords(inputFlowRecords)
+ val flows : DataFrame = cleanFlowRecords(inputFlowRecords)
+
+
+ logger.info("Fitting probabilistic model to data")
+ val model =
+ FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*), config.topicCount)
+
+ logger.info("Identifying outliers")
+ val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
- val scoredFlowRecords = detectFlowAnomalies(cleanFlowRecords, config, sparkContext, sqlContext, logger)
val filteredFlowRecords = filterScoredFlowRecords(scoredFlowRecords, config.threshold)
@@ -61,39 +68,18 @@ object FlowSuspiciousConnectsAnalysis {
val invalidFlowRecords = filterAndSelectInvalidFlowRecords(inputFlowRecords)
dataValidation.showAndSaveInvalidRecords(invalidFlowRecords, config.hdfsScoredConnect, logger)
+ outputFlowRecords
}
- /**
- * Identify anomalous netflow log entries in in the provided data frame.
- *
- * @param data Data frame of netflow entries
- * @param config
- * @param sparkContext
- * @param sqlContext
- * @param logger
- * @return
- */
- def detectFlowAnomalies(data: DataFrame,
- config: SuspiciousConnectsConfig,
- sparkContext: SparkContext,
- sqlContext: SQLContext,
- logger: Logger): DataFrame = {
-
- logger.info("Fitting probabilistic model to data")
- val model =
- FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, data, config.topicCount)
- logger.info("Identifying outliers")
- model.score(sparkContext, sqlContext, data)
- }
/**
*
* @param inputFlowRecords raw flow records
* @return
*/
- def filterAndSelectCleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
+ def cleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
val cleanFlowRecordsFilter = inputFlowRecords(Hour).between(0, 23) &&
inputFlowRecords(Minute).between(0, 59) &&
@@ -108,8 +94,6 @@ object FlowSuspiciousConnectsAnalysis {
inputFlowRecords
.filter(cleanFlowRecordsFilter)
- .select(InSchema: _*)
-
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/f5f57851/spot-ml/src/scripts/TestingFlow.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/scripts/TestingFlow.scala b/spot-ml/src/scripts/TestingFlow.scala
new file mode 100644
index 0000000..9876945
--- /dev/null
+++ b/spot-ml/src/scripts/TestingFlow.scala
@@ -0,0 +1,34 @@
+
+
+import org.apache.spot.netflow._
+import org.apache.spot.SuspiciousConnectsArgumentParser._
+import org.apache.log4j.{Level, LogManager, Logger}
+val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
+
+val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
+val FeedbackFile = ""
+val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11/baseline.scored.pqt"
+
+
+val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
+
+val config = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = InputParquetFile,
+ feedbackFile = FeedbackFile,
+ duplicationFactor = 1000,
+ topicCount = 20,
+ userDomain = "intel",
+ hdfsScoredConnect = ScoredParquetFile,
+ threshold = 1.0d,
+ maxResults = -1,
+ outputDelimiter = "\t",
+ ldaPRGSeed = Some(0xdeadbeef),
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+
+val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sc, sqlContext, logger, inputDataFrame).cache()
+
+// there are two big tasks .... rank the attacks and calculate the area under the ROC curve
+
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/f5f57851/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index 29e3363..a92d052 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.netflow.FlowSchema._
+import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
@@ -73,12 +74,17 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
typicalRecord, typicalRecord, typicalRecord, typicalRecord))
- val scoredData: DataFrame = FlowSuspiciousConnectsAnalysis.detectFlowAnomalies(data,
- testConfig,
- sparkContext,
- sqlContext,
- logger)
+ import FlowSuspiciousConnectsAnalysis.{cleanFlowRecords, InSchema}
+ val flows : DataFrame = cleanFlowRecords(data)
+
+
+ logger.info("Fitting probabilistic model to data")
+ val model =
+ FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, testConfig, flows.select(InSchema: _*), testConfig.topicCount)
+
+ logger.info("Identifying outliers")
+ val scoredData = model.score(sparkContext, sqlContext, flows)
val anomalyScore = scoredData.filter(scoredData(Hour) === 0).first().getAs[Double](Score)
@@ -100,8 +106,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
"filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
- val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis
- .filterAndSelectCleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
+ val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
cleanedFlowRecords.count should be(5)
cleanedFlowRecords.schema.size should be(17)
[3/8] incubator-spot git commit: flow_cleaning
Posted by ra...@apache.org.
flow_cleaning
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/3ba71e9d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/3ba71e9d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/3ba71e9d
Branch: refs/heads/master
Commit: 3ba71e9d8181f4866d3c08513dff6462f1d77837
Parents: f5f5785
Author: nlsegerl <na...@intel.com>
Authored: Mon May 22 09:58:50 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Mon May 22 09:58:50 2017 -0700
----------------------------------------------------------------------
.../org/apache/spot/natetest/FlowTune.scala | 90 +++++++++++++++
.../apache/spot/netflow/FlowColumnIndex.scala | 39 -------
.../FlowSuspiciousConnectsAnalysis.scala | 2 +-
.../model/FlowSuspiciousConnectsModel.scala | 53 ++++++---
spot-ml/src/scripts/TestingFlow.scala | 43 +++++++-
.../org/apache/spot/netflow/FlowRecord.scala | 22 ++++
.../FlowSuspiciousConnectsAnalysisTest.scala | 29 +----
.../model/FlowSuspiciousConnectsModelTest.scala | 109 +++++++++++++++++++
8 files changed, 298 insertions(+), 89 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala b/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
new file mode 100644
index 0000000..49471cc
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
@@ -0,0 +1,90 @@
+package org.apache.spot.natetest
+
+
+import org.apache.spot.netflow._
+import org.apache.spot.SuspiciousConnectsArgumentParser._
+import org.apache.log4j.{Level, LogManager, Logger}
+import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
+import org.apache.spark.sql.types.{LongType, StructField, StructType}
+import org.apache.spark.sql.{Row, SQLContext}
+import org.apache.spark.{SparkConf, SparkContext}
+/**
+ * Top level entrypoint to execute suspicious connections analysis on network data.
+ * Supported analyses:
+ * flow : netflow data
+ * dns : DNS server logs
+ * proxy : proxy server logs
+ */
+
+object FlowTune {
+
+ /**
+ * Execute suspicious connections analysis on network data.
+ *
+ * @param args Command line arguments.
+ */
+ def main(args: Array[String]) {
+
+ val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
+
+ Logger.getLogger("org").setLevel(Level.WARN)
+ Logger.getLogger("akka").setLevel(Level.OFF)
+
+ val sparkConfig = new SparkConf().setAppName("Netflow tuning")
+ val sparkContext = new SparkContext(sparkConfig)
+ val sqlContext = new SQLContext(sparkContext)
+
+ val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
+ val FeedbackFile = ""
+ val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11.baseline.scored.pqt"
+
+
+ val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
+
+ val config = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = InputParquetFile,
+ feedbackFile = FeedbackFile,
+ duplicationFactor = 1000,
+ topicCount = 20,
+ userDomain = "intel",
+ hdfsScoredConnect = ScoredParquetFile,
+ threshold = 1.0d,
+ maxResults = -1,
+ outputDelimiter = "\t",
+ ldaPRGSeed = Some(0xdeadbeef),
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+
+ val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sparkContext, sqlContext, logger, inputDataFrame).cache()
+
+ println("let me show you ten: ")
+ scoredDF.show(10)
+
+ // let's do the ROC first
+
+ val E = new BinaryClassificationMetrics(scoredDF.select("score", "is_attack").rdd.map({case (r: Row) => (r.getAs[Double](0), r.getAs[Double](1))}))
+
+ val auc = E.areaUnderROC()
+
+ println("Here's your area under the ROC curve: " + auc)
+
+ System.exit(0)
+
+ println("and now we have a dataframe with all of the ranked attacks")
+
+ val newSchema = StructType(scoredDF.schema.fields ++ Array(StructField("rank", LongType, false)))
+
+ val newRowsRDD = scoredDF.rdd.zipWithIndex().map({case (r,i) => Row.fromSeq(r.toSeq ++ Array[Any](i))})
+
+ val indexedAll = sqlContext.createDataFrame(newRowsRDD, newSchema)
+
+ val indexedAttacks = indexedAll.filter(indexedAll("is_attack") === 1)
+
+ indexedAttacks.show(20)
+ // i guess I will need to save this somewhere
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
deleted file mode 100644
index 173ab2d..0000000
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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 org.apache.spot.netflow
-
-object FlowColumnIndex extends Enumeration {
- val HOUR = 4
- val MINUTE = 5
- val SECOND = 6
- val SOURCEIP = 8
- val DESTIP = 9
- val SOURCEPORT = 10
- val DESTPORT = 11
- val IPKT = 16
- val IBYT = 17
- val NUMTIME = 27
- val IBYTBIN = 28
- val IPKTYBIN = 29
- val TIMEBIN = 30
- val PORTWORD = 31
- val IPPAIR = 32
- val SOURCEWORD = 33
- val DESTWORD = 34
- }
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index 4e97be1..3360f2b 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -46,7 +46,7 @@ object FlowSuspiciousConnectsAnalysis {
logger.info("Fitting probabilistic model to data")
val model =
- FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*), config.topicCount)
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*))
logger.info("Identifying outliers")
val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index 95dca42..28b0457 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -155,12 +155,30 @@ object FlowSuspiciousConnectsModel {
val ModelColumns = ModelSchema.fieldNames.toList.map(col)
- def trainNewModel(sparkContext: SparkContext,
- sqlContext: SQLContext,
- logger: Logger,
- config: SuspiciousConnectsConfig,
- inputRecords: DataFrame,
- topicCount: Int): FlowSuspiciousConnectsModel = {
+
+ def cleanData(flows: DataFrame): DataFrame = {
+
+ val legalFlowsFilter = flows(Hour).between(0, 23) &&
+ flows(Minute).between(0, 59) &&
+ flows(Second).between(0, 59) &&
+ flows(TimeReceived).isNotNull &&
+ flows(SourceIP).isNotNull &&
+ flows(DestinationIP).isNotNull &&
+ flows(SourcePort).isNotNull &&
+ flows(DestinationPort).isNotNull &&
+ flows(Ibyt).isNotNull &&
+ flows(Ipkt).isNotNull
+
+ flows.filter(legalFlowsFilter)
+ }
+
+
+ def trainModel(sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger,
+ config: SuspiciousConnectsConfig,
+ inputRecords: DataFrame): FlowSuspiciousConnectsModel = {
+
logger.info("Training netflow suspicious connects model from " + config.inputPath)
@@ -281,17 +299,18 @@ object FlowSuspiciousConnectsModel {
.map({ case ((ip, word), count) => SpotLDAInput(ip, word, count) })
- val SpotLDAOutput(ipToTopicMix, wordToPerTopicProb) = SpotLDAWrapper.runLDA(sparkContext,
- sqlContext,
- ipWordCounts,
- config.topicCount,
- logger,
- config.ldaPRGSeed,
- config.ldaAlpha,
- config.ldaBeta,
- config.ldaMaxiterations)
-
- new FlowSuspiciousConnectsModel(topicCount,
+ val SpotLDAOutput(ipToTopicMix, wordToPerTopicProb) =
+ SpotLDAWrapper.runLDA(sparkContext,
+ sqlContext,
+ ipWordCounts,
+ config.topicCount,
+ logger,
+ config.ldaPRGSeed,
+ config.ldaAlpha,
+ config.ldaBeta,
+ config.ldaMaxiterations)
+
+ new FlowSuspiciousConnectsModel(config.topicCount,
ipToTopicMix,
wordToPerTopicProb,
config.useProtocol,
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/scripts/TestingFlow.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/scripts/TestingFlow.scala b/spot-ml/src/scripts/TestingFlow.scala
index 9876945..a713fba 100644
--- a/spot-ml/src/scripts/TestingFlow.scala
+++ b/spot-ml/src/scripts/TestingFlow.scala
@@ -1,13 +1,22 @@
import org.apache.spot.netflow._
+import org.apache.spot.netflow.FlowSchema._
import org.apache.spot.SuspiciousConnectsArgumentParser._
import org.apache.log4j.{Level, LogManager, Logger}
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
+
val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
-val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
+Logger.getLogger("org").setLevel(Level.WARN)
+Logger.getLogger("akka").setLevel(Level.OFF)
+
+val InputParquetFile = "/user/duxbury/flow/hive/y=2016/m=05/d=05"
+//val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
val FeedbackFile = ""
-val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11/baseline.scored.pqt"
+val ScoredParquetFile = "/user/nlsegerl/test/2016.05.05.baseline.scored.pqt"
val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
@@ -16,19 +25,41 @@ val config = SuspiciousConnectsConfig(analysis = "flow",
inputPath = InputParquetFile,
feedbackFile = FeedbackFile,
duplicationFactor = 1000,
- topicCount = 20,
+ topicCount = 10,
userDomain = "intel",
hdfsScoredConnect = ScoredParquetFile,
threshold = 1.0d,
maxResults = -1,
outputDelimiter = "\t",
ldaPRGSeed = Some(0xdeadbeef),
- ldaMaxiterations = 20,
+ ldaMaxiterations = 5,
ldaAlpha = 1.02,
ldaBeta = 1.001)
-val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sc, sqlContext, logger, inputDataFrame).cache()
-// there are two big tasks .... rank the attacks and calculate the area under the ROC curve
+logger.info("Starting flow suspicious connects analysis.")
+
+val cleanFlows : DataFrame = FlowSuspiciousConnectsModel.cleanData(inputDataFrame) // ? should we cache here?
+
+logger.info("Fitting model to data")
+
+val model = FlowSuspiciousConnectsModel.trainModel(sc, sqlContext, logger, config, cleanFlows)
+
+logger.info("Assigning scores to flow records")
+
+val scoredFlowRecords = model.score(sc, sqlContext, cleanFlows)
+
+logger.info("Selecting most suspicious flow records")
+
+val orderedFlowRecords = scoredFlowRecords.filter(scoredFlowRecords(Score) < 0.0001).orderBy(Score)
+
+
+
+orderedFlowRecords.show(10)
+
+
+val df = orderedFlowRecords
+
+
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
new file mode 100644
index 0000000..8349c9e
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
@@ -0,0 +1,22 @@
+package org.apache.spot.netflow
+
+/**
+ * Created by nlsegerl on 5/16/17.
+ */
+case class FlowRecord(treceived: String,
+ tryear: Int,
+ trmonth: Int,
+ trday: Int,
+ trhour: Int,
+ trminute: Int,
+ trsec: Int,
+ tdur: Float,
+ sip: String,
+ dip: String,
+ sport: Int,
+ dport: Int,
+ proto: String,
+ ipkt: Int,
+ ibyt: Int,
+ opkt: Int,
+ obyt: Int)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index a92d052..c16fb13 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -26,23 +26,7 @@ import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
-case class FlowRecord(treceived: String,
- tryear: Int,
- trmonth: Int,
- trday: Int,
- trhour: Int,
- trminute: Int,
- trsec: Int,
- tdur: Float,
- sip: String,
- dip: String,
- sport: Int,
- dport: Int,
- proto: String,
- ipkt: Int,
- ibyt: Int,
- opkt: Int,
- obyt: Int)
+
class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Matchers {
@@ -75,13 +59,12 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- import FlowSuspiciousConnectsAnalysis.{cleanFlowRecords, InSchema}
- val flows : DataFrame = cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsModel.cleanData(data)
logger.info("Fitting probabilistic model to data")
val model =
- FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, testConfig, flows.select(InSchema: _*), testConfig.topicCount)
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, testConfig, flows)
logger.info("Identifying outliers")
val scoredData = model.score(sparkContext, sqlContext, flows)
@@ -104,13 +87,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
- "filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
-
- val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
- cleanedFlowRecords.count should be(5)
- cleanedFlowRecords.schema.size should be(17)
- }
"filterAndSelectInvalidFlowRecords" should "return invalid records" in {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
new file mode 100644
index 0000000..0f52713
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -0,0 +1,109 @@
+package org.apache.spot.netflow.model
+
+import org.apache.log4j.{Level, LogManager}
+import org.apache.spot.netflow.FlowRecord
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{DataFrame, Row, SQLContext}
+import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
+import org.apache.spot.netflow.FlowSchema._
+import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
+import org.apache.spot.testutils.TestingSparkContextFlatSpec
+import org.scalatest.Matchers
+
+class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with Matchers {
+
+ val testConfig = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = "",
+ feedbackFile = "",
+ duplicationFactor = 1,
+ topicCount = 20,
+ hdfsScoredConnect = "",
+ threshold = 1.0d,
+ maxResults = 1000,
+ outputDelimiter = "\t",
+ ldaPRGSeed = None,
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+ "filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
+
+ val cleanedFlowRecords = FlowSuspiciousConnectsModel.cleanData(testFlowRecords.inputFlowRecordsDF)
+
+ cleanedFlowRecords.count should be(5)
+ cleanedFlowRecords.schema.size should be(17)
+ }
+
+ def testFlowRecords = new {
+ val sqlContext = new SQLContext(sparkContext)
+
+ val inputFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 24, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 60, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 60, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq(null, 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, null, "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", null, 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", null, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, null, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", null, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, null, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, null, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, null),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0))
+ .map(row => Row.fromSeq(row))))
+
+ val inputFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField))
+
+ val inputFlowRecordsDF = sqlContext.createDataFrame(inputFlowRecordsRDD, inputFlowRecordsSchema)
+
+ val scoredFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, -1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0000005),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.05),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0001))
+ .map(row => Row.fromSeq(row))))
+
+ val scoredFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField,
+ ScoreField))
+
+ val scoredFlowRecordsDF = sqlContext.createDataFrame(scoredFlowRecordsRDD, scoredFlowRecordsSchema)
+ }
+}
[7/8] incubator-spot git commit: flow_tuning
Posted by ra...@apache.org.
flow_tuning
incorporated code review feedback
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/4004ede7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/4004ede7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/4004ede7
Branch: refs/heads/master
Commit: 4004ede754a594c401c5b14264c1e6c8484fa880
Parents: f18bd55
Author: nlsegerl <na...@intel.com>
Authored: Fri Jun 2 14:24:50 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Fri Jun 2 14:24:50 2017 -0700
----------------------------------------------------------------------
.../FlowSuspiciousConnectsAnalysis.scala | 5 +-
.../apache/spot/netflow/FlowWordCreator.scala | 7 +--
.../model/FlowSuspiciousConnectsModel.scala | 15 -----
.../org/apache/spot/utilities/MathUtils.scala | 8 +++
spot-ml/src/scripts/TestingFlow.scala | 65 --------------------
.../FlowSuspiciousConnectsAnalysisTest.scala | 4 +-
.../spot/netflow/FlowWordCreatorTest.scala | 53 +++++++++++-----
.../model/FlowSuspiciousConnectsModelTest.scala | 4 +-
8 files changed, 55 insertions(+), 106 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index 3360f2b..e972bf4 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -37,7 +37,7 @@ object FlowSuspiciousConnectsAnalysis {
def run(config: SuspiciousConnectsConfig, sparkContext: SparkContext, sqlContext: SQLContext, logger: Logger,
- inputFlowRecords: DataFrame) : DataFrame = {
+ inputFlowRecords: DataFrame) = {
logger.info("Starting flow suspicious connects analysis.")
@@ -46,7 +46,7 @@ object FlowSuspiciousConnectsAnalysis {
logger.info("Fitting probabilistic model to data")
val model =
- FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*))
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows)
logger.info("Identifying outliers")
val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
@@ -68,7 +68,6 @@ object FlowSuspiciousConnectsAnalysis {
val invalidFlowRecords = filterAndSelectInvalidFlowRecords(inputFlowRecords)
dataValidation.showAndSaveInvalidRecords(invalidFlowRecords, config.hdfsScoredConnect, logger)
- outputFlowRecords
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
index 119eaa8..4365ecd 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
@@ -19,6 +19,7 @@ package org.apache.spot.netflow
import org.apache.spark.sql.functions._
import org.apache.spot.utilities.data.validation.InvalidDataHandler
+import org.apache.spot.utilities.MathUtils.ceilLog2
import scala.util.{Success, Try}
@@ -83,11 +84,9 @@ object FlowWordCreator extends Serializable {
def flowWords(hour: Int, srcPort: Int, dstPort: Int, protocol: String, ibyt: Long, ipkt: Long): FlowWords = {
Try {
- val lnOf2 = scala.math.log(2) // natural log of 2
- val ibytBin: Long =
- scala.math.ceil(scala.math.log(ibyt) / lnOf2).toLong // 0 values should never ever happen
- val ipktBin: Long = scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
+ val ibytBin = ceilLog2(ibyt + 1)
+ val ipktBin = ceilLog2(ipkt + 1)
val LowToLowPortEncoding = 111111
val HighToHighPortEncoding = 333333
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index 17da8a1..25ff7c8 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -132,21 +132,6 @@ object FlowSuspiciousConnectsModel {
val ModelColumns = ModelSchema.fieldNames.toList.map(col)
- def cleanData(flows: DataFrame): DataFrame = {
-
- val legalFlowsFilter = flows(Hour).between(0, 23) &&
- flows(TimeReceived).isNotNull &&
- flows(SourceIP).isNotNull &&
- flows(DestinationIP).isNotNull &&
- flows(SourcePort).isNotNull &&
- flows(DestinationPort).isNotNull &&
- flows(Ibyt).isNotNull &&
- flows(Ipkt).isNotNull
-
- flows.filter(legalFlowsFilter)
- }
-
-
def trainModel(sparkContext: SparkContext,
sqlContext: SQLContext,
logger: Logger,
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/main/scala/org/apache/spot/utilities/MathUtils.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/MathUtils.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/MathUtils.scala
index e5f30b7..b0f2388 100644
--- a/spot-ml/src/main/scala/org/apache/spot/utilities/MathUtils.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/MathUtils.scala
@@ -28,4 +28,12 @@ object MathUtils {
* @return y rounded down to an integer
*/
def logBaseXInt(x: Double, base: Int): Int = if (x == 0) 0 else (log10(x) / log10(base)).toInt
+
+ /**
+ * Returns the ceiling of the logarithm base 2 of the incoming double.
+ *
+ * @param x A double.
+ * @return Integer ceiling of logarithm base-2 of x.
+ */
+ def ceilLog2(x: Double) : Int = Math.ceil(log10(x) / log10(2d)).toInt
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/scripts/TestingFlow.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/scripts/TestingFlow.scala b/spot-ml/src/scripts/TestingFlow.scala
deleted file mode 100644
index a713fba..0000000
--- a/spot-ml/src/scripts/TestingFlow.scala
+++ /dev/null
@@ -1,65 +0,0 @@
-
-
-import org.apache.spot.netflow._
-import org.apache.spot.netflow.FlowSchema._
-import org.apache.spot.SuspiciousConnectsArgumentParser._
-import org.apache.log4j.{Level, LogManager, Logger}
-import org.apache.spark.{SparkConf, SparkContext}
-import org.apache.spark.sql.{DataFrame, SQLContext}
-import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
-
-val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
-
-Logger.getLogger("org").setLevel(Level.WARN)
-Logger.getLogger("akka").setLevel(Level.OFF)
-
-val InputParquetFile = "/user/duxbury/flow/hive/y=2016/m=05/d=05"
-//val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
-val FeedbackFile = ""
-val ScoredParquetFile = "/user/nlsegerl/test/2016.05.05.baseline.scored.pqt"
-
-
-val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
-
-val config = SuspiciousConnectsConfig(analysis = "flow",
- inputPath = InputParquetFile,
- feedbackFile = FeedbackFile,
- duplicationFactor = 1000,
- topicCount = 10,
- userDomain = "intel",
- hdfsScoredConnect = ScoredParquetFile,
- threshold = 1.0d,
- maxResults = -1,
- outputDelimiter = "\t",
- ldaPRGSeed = Some(0xdeadbeef),
- ldaMaxiterations = 5,
- ldaAlpha = 1.02,
- ldaBeta = 1.001)
-
-
-
-logger.info("Starting flow suspicious connects analysis.")
-
-val cleanFlows : DataFrame = FlowSuspiciousConnectsModel.cleanData(inputDataFrame) // ? should we cache here?
-
-logger.info("Fitting model to data")
-
-val model = FlowSuspiciousConnectsModel.trainModel(sc, sqlContext, logger, config, cleanFlows)
-
-logger.info("Assigning scores to flow records")
-
-val scoredFlowRecords = model.score(sc, sqlContext, cleanFlows)
-
-logger.info("Selecting most suspicious flow records")
-
-val orderedFlowRecords = scoredFlowRecords.filter(scoredFlowRecords(Score) < 0.0001).orderBy(Score)
-
-
-
-orderedFlowRecords.show(10)
-
-
-val df = orderedFlowRecords
-
-
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index afded0a..2813f43 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -59,7 +59,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsModel.cleanData(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
@@ -120,7 +120,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsModel.cleanData(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
index 583728c..1fc5266 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
@@ -63,7 +63,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 3. Test when sip is less than dip and sip is not 0 and dport and sport are > 1024 +
- it should "create word with ip_pair as sourceIp-destIp, port is 333333 and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as sourceIp-destIp, port is 333333 and both words direction is positive" in {
val srcPort = 8392
val dstPort = 9874
@@ -88,7 +88,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 5. Test when sip is less than dip and sip is not 0 and sport is 0 but dport is not +
- it should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 II" in {
+ it should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 (case 2)" in {
val srcPort = 0
val dstPort = 43
@@ -103,7 +103,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 6. Test when sip is less than dip and sip is not 0 and sport and dport are less or equal than 1024 +
- it should "create word with ip_pair as sourceIp-destIp, port is 111111 and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as sourceIp-destIp, port is 111111 and both words in positive direction" in {
val srcPort = 1024
val dstPort = 80
@@ -116,7 +116,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 7. Test when sip is less than dip and sip is not 0 and sport and dport are 0+
- it should "create word with ip_pair as sourceIp-destIp, port is max(0,0) and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as sourceIp-destIp, port is max(0,0) and both words in positive direction" in {
val srcPort = 0
val dstPort = 0
@@ -156,7 +156,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 10. Test when sip is not less than dip and dport and sport are > 1024 +
- it should "create word with ip_pair as destIp-sourceIp, port is 333333 and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is 333333 and both words direction is positive" in {
val srcPort = 2354
val dstPort = 2435
@@ -169,7 +169,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 11. Test when sip is not less than dip and dport is 0 but sport is not +
- it should "create word with ip_pair as destIp-sourceIp, port is sport and src_word direction is -1 II" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is sport and src_word direction is -1 (Case 2)" in {
val srcPort = 80
val dstPort = 0
@@ -182,7 +182,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 12. Test when sip is not less than dip and sport is 0 but dport is not +
- it should "create word with ip_pair as destIp-sourceIp, port is dport and dest_word direction is -1 II" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is dport and dest_word direction is -1 (case 2)" in {
val srcPort = 0
val dstPort = 2435
@@ -195,7 +195,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 13. Test when sip is not less than dip and sport and dport are less or equal than 1024
- it should "create word with ip_pair as destIp-sourceIp, port 111111 and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as destIp-sourceIp, port 111111 and both words direction is positive" in {
val srcPort = 80
val dstPort = 1024
@@ -208,7 +208,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
// 14. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing)" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is positive, protocol UDP" in {
val srcPort = 0
val dstPort = 0
@@ -220,8 +220,8 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
srcWord shouldBe "0_UDP_12_8_2"
}
- // 15. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
- "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 and including protocol" in {
+ // 15. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 and protocol is TCP" in {
val srcPort = 2132
val dstPort = 23
@@ -237,7 +237,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
// 16. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) and including protocol" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is positive, protocol TCP" in {
val srcPort = 0
val dstPort = 0
val protocol = "TCP"
@@ -249,8 +249,8 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
- // 17. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
- "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with protocol + modified bins" in {
+ // 17. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0
+ "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 and protocol is UDP" in {
val srcPort = 2132
val dstPort = 23
@@ -265,7 +265,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
// 18. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with protocol + modified bins" in {
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is positive " in {
val srcPort = 0
val dstPort = 0
val protocol = "TCP"
@@ -277,9 +277,32 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
}
+ // 19. Test when sip is not less than dip and sport and dport are 0 and bytecount is 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is positive with zero bytes" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val ibyts = 0
+ val FlowWords(srcWord, dstWord) =
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
+ dstWord shouldBe "0_TCP_12_0_2"
+ srcWord shouldBe "0_TCP_12_0_2"
+ }
+ // 20. Test when sip is not less than dip and sport and dport are 0 and packet count is 0
+ it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is positive with protocol with zero packets" in {
+ val srcPort = 0
+ val dstPort = 0
+ val protocol = "TCP"
+ val ipkts = 0
+ val FlowWords(srcWord, dstWord) =
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
+
+ dstWord shouldBe "0_TCP_12_8_0"
+ srcWord shouldBe "0_TCP_12_8_0"
+ }
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/4004ede7/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
index 50fe696..035b962 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -1,7 +1,7 @@
package org.apache.spot.netflow.model
import org.apache.log4j.{Level, LogManager}
-import org.apache.spot.netflow.FlowRecord
+import org.apache.spot.netflow.{FlowRecord, FlowSuspiciousConnectsAnalysis}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
@@ -28,7 +28,7 @@ class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with M
"filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
- val cleanedFlowRecords = FlowSuspiciousConnectsModel.cleanData(testFlowRecords.inputFlowRecordsDF)
+ val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
cleanedFlowRecords.count should be(7)
cleanedFlowRecords.schema.size should be(17)
[8/8] incubator-spot git commit: flow_tuning
Posted by ra...@apache.org.
flow_tuning
reverted some changes to defer them for a larger overhaul
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/c0f80c56
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/c0f80c56
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/c0f80c56
Branch: refs/heads/master
Commit: c0f80c56ea3f8cae666c3875fb87b3b175fdd915
Parents: 4004ede
Author: nlsegerl <na...@intel.com>
Authored: Tue Jun 6 10:26:20 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Tue Jun 6 10:26:20 2017 -0700
----------------------------------------------------------------------
.../FlowSuspiciousConnectsAnalysis.scala | 34 ++++++++++++++------
.../FlowSuspiciousConnectsAnalysisTest.scala | 4 +--
.../model/FlowSuspiciousConnectsModelTest.scala | 8 ++---
3 files changed, 30 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index e972bf4..7469e13 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -41,15 +41,11 @@ object FlowSuspiciousConnectsAnalysis {
logger.info("Starting flow suspicious connects analysis.")
- val flows : DataFrame = cleanFlowRecords(inputFlowRecords)
+ val flows : DataFrame = filterAndSelectCleanFlowRecords(inputFlowRecords)
- logger.info("Fitting probabilistic model to data")
- val model =
- FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows)
-
logger.info("Identifying outliers")
- val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
+ val scoredFlowRecords = detectFlowAnomalies(flows, config, sparkContext, sqlContext, logger)
val filteredFlowRecords = filterScoredFlowRecords(scoredFlowRecords, config.threshold)
@@ -71,6 +67,27 @@ object FlowSuspiciousConnectsAnalysis {
}
+ /**
+ * Identify anomalous netflow log entries in in the provided data frame.
+ *
+ * @param data Data frame of netflow entries
+ * @param config
+ * @param sparkContext
+ * @param sqlContext
+ * @param logger
+ * @return
+ */
+ def detectFlowAnomalies(data: DataFrame,
+ config: SuspiciousConnectsConfig,
+ sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger): DataFrame = {
+ logger.info("Fitting probabilistic model to data")
+ val model =
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, data)
+ logger.info("Identifying outliers")
+ model.score(sparkContext, sqlContext, data)
+ }
/**
@@ -78,8 +95,7 @@ object FlowSuspiciousConnectsAnalysis {
* @param inputFlowRecords raw flow records
* @return
*/
- def cleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
-
+ def filterAndSelectCleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
val cleanFlowRecordsFilter = inputFlowRecords(Hour).between(0, 23) &&
inputFlowRecords(Minute).between(0, 59) &&
inputFlowRecords(Second).between(0, 59) &&
@@ -90,9 +106,9 @@ object FlowSuspiciousConnectsAnalysis {
inputFlowRecords(DestinationPort).isNotNull &&
inputFlowRecords(Ibyt).isNotNull &&
inputFlowRecords(Ipkt).isNotNull
-
inputFlowRecords
.filter(cleanFlowRecordsFilter)
+ .select(InSchema: _*)
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index 2813f43..d13abc3 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -59,7 +59,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
@@ -120,7 +120,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
index 035b962..82c3813 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -1,12 +1,10 @@
package org.apache.spot.netflow.model
-import org.apache.log4j.{Level, LogManager}
-import org.apache.spot.netflow.{FlowRecord, FlowSuspiciousConnectsAnalysis}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.{DataFrame, Row, SQLContext}
+import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.netflow.FlowSchema._
-import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
+import org.apache.spot.netflow.FlowSuspiciousConnectsAnalysis
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
@@ -28,7 +26,7 @@ class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with M
"filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
- val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
+ val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
cleanedFlowRecords.count should be(7)
cleanedFlowRecords.schema.size should be(17)
[5/8] incubator-spot git commit: flow_tuning
Posted by ra...@apache.org.
flow_tuning
removed minute and second from the model columns since they are no longer being used
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/e093531a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/e093531a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/e093531a
Branch: refs/heads/master
Commit: e093531a09374443870c695dcf344d2bf548db89
Parents: 78b0090
Author: nlsegerl <na...@intel.com>
Authored: Thu Jun 1 16:57:21 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Thu Jun 1 16:57:21 2017 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/sql/WideUDFs.scala | 19 +++++++++
.../apache/spot/netflow/FlowWordCreator.scala | 41 +++++++-------------
.../spot/netflow/model/FlowScoreFunction.scala | 8 +---
.../model/FlowSuspiciousConnectsModel.scala | 11 +-----
.../spot/netflow/FlowWordCreatorTest.scala | 36 ++++++++---------
.../model/FlowSuspiciousConnectsModelTest.scala | 6 +--
6 files changed, 57 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala b/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
index 28ecdc2..fca37e2 100644
--- a/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
+++ b/spot-ml/src/main/scala/org/apache/spark/sql/WideUDFs.scala
@@ -24,6 +24,25 @@ import scala.reflect.runtime.universe.{TypeTag, typeTag}
import scala.util.Try
object WideUDFs {
+
+ def udf[RT: TypeTag, A1: TypeTag, A2: TypeTag, A3: TypeTag, A4: TypeTag, A5: TypeTag, A6: TypeTag,
+ A7: TypeTag, A8: TypeTag, A9: TypeTag, A10: TypeTag, A11: TypeTag](f: Function10[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, RT]):
+ UserDefinedFunction = {
+ val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A2]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A3]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A4]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A5]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A6]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A7]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A8]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A9]).dataType
+ :: ScalaReflection.schemaFor(typeTag[A10]).dataType
+ :: Nil).getOrElse(Nil)
+ UserDefinedFunction(f, ScalaReflection.schemaFor(typeTag[RT]).dataType, inputTypes)
+ }
+
+
/**
* Defines a user-defined function of 11 arguments as user-defined function (UDF).
* The data types are automatically inferred based on the function's signature.
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
index f5e1a4b..119eaa8 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
@@ -18,7 +18,6 @@
package org.apache.spot.netflow
import org.apache.spark.sql.functions._
-import org.apache.spot.utilities.Quantiles
import org.apache.spot.utilities.data.validation.InvalidDataHandler
import scala.util.{Success, Try}
@@ -45,8 +44,6 @@ object FlowWordCreator extends Serializable {
* @return String "word" summarizing a netflow connection.
*/
def srcWordUDF = udf((hour: Int,
- minute: Int,
- second: Int,
srcIP: String,
dstIP: String,
srcPort: Int,
@@ -54,7 +51,7 @@ object FlowWordCreator extends Serializable {
protocol: String,
ibyt: Long,
ipkt: Long) =>
- flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyt, ipkt).srcWord)
+ flowWords(hour, srcPort, dstPort, protocol, ibyt, ipkt).srcWord)
/**
@@ -63,43 +60,34 @@ object FlowWordCreator extends Serializable {
* @return String "word" summarizing a netflow connection.
*/
def dstWordUDF = udf((hour: Int,
- minute: Int,
- second: Int,
srcIP: String,
dstIP: String,
srcPort: Int,
dstPort: Int,
- protocol: String,
+ protocol: String,
ibyt: Long,
ipkt: Long) =>
- flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyt, ipkt).dstWord)
+ flowWords(hour, srcPort, dstPort, protocol, ibyt, ipkt).dstWord)
/**
* Calculate the source and destination words summarizing a netflow record.
*
* @param hour
- * @param minute
- * @param second
* @param srcPort
* @param dstPort
* @param ipkt
* @param ibyt
* @return [[FlowWords]] containing source and destination words.
*/
- def flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol: String, ibyt: Long, ipkt: Long): FlowWords = {
+ def flowWords(hour: Int, srcPort: Int, dstPort: Int, protocol: String, ibyt: Long, ipkt: Long): FlowWords = {
Try {
- val timeOfDay: Double = hour.toDouble + minute.toDouble / 60 + second.toDouble / 3600
-
- val timeBin = hour
-
-
val lnOf2 = scala.math.log(2) // natural log of 2
- val ibytBin : Long =
- scala.math.ceil(scala.math.log(ibyt) / lnOf2).toLong // 0 values should never ever happen
+ val ibytBin: Long =
+ scala.math.ceil(scala.math.log(ibyt) / lnOf2).toLong // 0 values should never ever happen
- val ipktBin : Long = scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
+ val ipktBin: Long = scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
val LowToLowPortEncoding = 111111
val HighToHighPortEncoding = 333333
@@ -109,32 +97,32 @@ object FlowWordCreator extends Serializable {
if (dstPort == 0 && srcPort == 0) {
- val baseWord = Array("0", proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array("0", proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
} else if (dstPort == 0 && srcPort > 0) {
- val baseWord = Array(srcPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(srcPort, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = "-1_" + baseWord, dstWord = baseWord)
} else if (srcPort == 0 && dstPort > 0) {
- val baseWord = Array(dstPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else if (srcPort <= 1024 && dstPort <= 1024) {
- val baseWord = Array(LowToLowPortEncoding, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(LowToLowPortEncoding, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
} else if (srcPort <= 1024 && dstPort > 1024) {
- val baseWord = Array(srcPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(srcPort, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = "-1_" + baseWord, dstWord = baseWord)
} else if (srcPort > 1024 && dstPort <= 1024) {
- val baseWord = Array(dstPort, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else {
@@ -142,8 +130,7 @@ object FlowWordCreator extends Serializable {
// this is the srcPort > 1024 && dstPort > 1024 case
-
- val baseWord = Array(HighToHighPortEncoding, proto, timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(HighToHighPortEncoding, proto, hour, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = baseWord)
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
index d567f0e..866eaa9 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
@@ -40,8 +40,6 @@ class FlowScoreFunction(topicCount: Int,
* and assign it the least of these two values.
*
* @param hour Hour of flow record.
- * @param minute Minute of flow record.
- * @param second Second of flow record.
* @param srcIP Source IP of flow record.
* @param dstIP Destination IP of flow record.
* @param srcPort Source port of flow record.
@@ -53,20 +51,18 @@ class FlowScoreFunction(topicCount: Int,
* @return Minium of probability of this word from the source IP and probability of this word from the dest IP.
*/
def score(hour: Int,
- minute: Int,
- second: Int,
srcIP: String,
dstIP: String,
srcPort: Int,
dstPort: Int,
protocol: String,
- ipkt: Long,
ibyt: Long,
+ ipkt: Long,
srcTopicMix: Seq[Double],
dstTopicMix: Seq[Double]): Double = {
- val FlowWords(srcWord, dstWord) = FlowWordCreator.flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol, ibyt: Long, ipkt: Long)
+ val FlowWords(srcWord, dstWord) = FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyt, ipkt)
val zeroProb = Array.fill(topicCount) {
0.0
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index e9707cf..17da8a1 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -29,7 +29,6 @@ import org.apache.spot.lda.SpotLDAWrapper.{SpotLDAInput, SpotLDAOutput}
import org.apache.spot.lda.SpotLDAWrapperSchema._
import org.apache.spot.netflow.FlowSchema._
import org.apache.spot.netflow.FlowWordCreator
-import org.apache.spot.utilities.Quantiles
import org.apache.spot.utilities.data.validation.InvalidDataHandler
import scala.util.{Failure, Success, Try}
@@ -86,20 +85,16 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
val scoringUDF = udf((hour: Int,
- minute: Int,
- second: Int,
srcIP: String,
dstIP: String,
srcPort: Int,
dstPort: Int,
protocol: String,
- ipkt: Long,
ibyt: Long,
+ ipkt: Long,
srcIpTopicMix: Seq[Double],
dstIpTopicMix: Seq[Double]) =>
scoreFunction.score(hour,
- minute,
- second,
srcIP,
dstIP,
srcPort,
@@ -126,8 +121,6 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
object FlowSuspiciousConnectsModel {
val ModelSchema = StructType(List(HourField,
- MinuteField,
- SecondField,
SourceIPField,
DestinationIPField,
SourcePortField,
@@ -142,8 +135,6 @@ object FlowSuspiciousConnectsModel {
def cleanData(flows: DataFrame): DataFrame = {
val legalFlowsFilter = flows(Hour).between(0, 23) &&
- flows(Minute).between(0, 59) &&
- flows(Second).between(0, 59) &&
flows(TimeReceived).isNotNull &&
flows(SourceIP).isNotNull &&
flows(DestinationIP).isNotNull &&
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
index c3df375..583728c 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
@@ -41,7 +41,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 23
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_23_UDP_12_8_2"
@@ -56,7 +56,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2132
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "23_UDP_12_8_2"
srcWord shouldBe "-1_23_UDP_12_8_2"
@@ -68,7 +68,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 9874
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "333333_UDP_12_8_2"
srcWord shouldBe "333333_UDP_12_8_2"
@@ -80,7 +80,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 0
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "80_UDP_12_8_2"
@@ -95,7 +95,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_43_UDP_12_8_2"
@@ -109,7 +109,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "111111_UDP_12_8_2"
srcWord shouldBe "111111_UDP_12_8_2"
@@ -122,7 +122,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "0_UDP_12_8_2"
srcWord shouldBe "0_UDP_12_8_2"
@@ -135,7 +135,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_43_UDP_12_8_2"
srcWord shouldBe "43_UDP_12_8_2"
@@ -148,7 +148,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2435
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "80_UDP_12_8_2"
srcWord shouldBe "-1_80_UDP_12_8_2"
@@ -162,7 +162,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "333333_UDP_12_8_2"
srcWord shouldBe "333333_UDP_12_8_2"
@@ -175,7 +175,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "80_UDP_12_8_2"
srcWord shouldBe "-1_80_UDP_12_8_2"
@@ -188,7 +188,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_2435_UDP_12_8_2"
srcWord shouldBe "2435_UDP_12_8_2"
@@ -201,7 +201,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "111111_UDP_12_8_2"
srcWord shouldBe "111111_UDP_12_8_2"
@@ -214,7 +214,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "0_UDP_12_8_2"
srcWord shouldBe "0_UDP_12_8_2"
@@ -227,7 +227,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val protocol = "TCP"
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_23_TCP_12_8_2"
@@ -242,7 +242,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 0
val protocol = "TCP"
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "0_TCP_12_8_2"
srcWord shouldBe "0_TCP_12_8_2"
@@ -255,7 +255,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 23
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_23_UDP_12_8_2"
@@ -270,7 +270,7 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 0
val protocol = "TCP"
val FlowWords(srcWord, dstWord) =
- FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "0_TCP_12_8_2"
srcWord shouldBe "0_TCP_12_8_2"
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/e093531a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
index 0f52713..50fe696 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -30,7 +30,7 @@ class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with M
val cleanedFlowRecords = FlowSuspiciousConnectsModel.cleanData(testFlowRecords.inputFlowRecordsDF)
- cleanedFlowRecords.count should be(5)
+ cleanedFlowRecords.count should be(7)
cleanedFlowRecords.schema.size should be(17)
}
@@ -39,8 +39,8 @@ class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with M
val inputFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
Seq("2016-05-05 13:54:58", 2016, 5, 5, 24, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
- Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 60, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
- Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 60, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 59, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 59, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
Seq(null, 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, null, "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", null, 1024, 80, "TCP", 39l, 12522l, 0, 0),
[4/8] incubator-spot git commit: flow_tuning
Posted by ra...@apache.org.
flow_tuning
Changed flow word creation logic.
* protocol is now part of the flow word
* time is binned by the hour
* byte count is binned by its logarithm
* packet count is binned by its logarithm
On our synthetic BP dataset, this led to substantial improvement in the model effectiveness.
It also removes two full dataest sorting passes from the model construction.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/78b0090a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/78b0090a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/78b0090a
Branch: refs/heads/master
Commit: 78b0090aa769ca9c7cbce2428f380286b11181c5
Parents: 3ba71e9
Author: nlsegerl <na...@intel.com>
Authored: Tue May 30 16:47:29 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Tue May 30 16:47:29 2017 -0700
----------------------------------------------------------------------
.../spot/SuspiciousConnectsArgumentParser.scala | 25 +--
.../org/apache/spot/natetest/FlowTune.scala | 90 ---------
.../apache/spot/netflow/FlowWordCreator.scala | 37 +---
.../spot/netflow/model/FlowScoreFunction.scala | 16 +-
.../model/FlowSuspiciousConnectsModel.scala | 112 +---------
.../FlowSuspiciousConnectsAnalysisTest.scala | 62 ++++++
.../spot/netflow/FlowWordCreatorTest.scala | 202 +++++--------------
7 files changed, 134 insertions(+), 410 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
index e42a610..b2e643a 100644
--- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala
@@ -36,14 +36,7 @@ object SuspiciousConnectsArgumentParser {
ldaPRGSeed: Option[Long] = None,
ldaMaxiterations: Int = 20,
ldaAlpha: Double = 1.02,
- ldaBeta: Double = 1.001,
- // flow tuning options
- // these are for the flow tuning branch only!
-
- flatBinTime: Boolean = false,
- expBinIBytes: Boolean = false,
- expBinIPkts: Boolean = false,
- useProtocol: Boolean = false)
+ ldaBeta: Double = 1.001)
val parser: scopt.OptionParser[SuspiciousConnectsConfig] = new scopt.OptionParser[SuspiciousConnectsConfig]("LDA") {
@@ -108,22 +101,6 @@ object SuspiciousConnectsArgumentParser {
text("topic concentration for lda, default 1.001")
- // these are for the flow tuning branch only!
- opt[Boolean]("hourbins").optional().
- action((x, c) => c.copy(flatBinTime = true)).
- text("use per-hour binning strategy for flow time of day")
-
- opt[Boolean]("expbinbytes").optional().
- action((x, c) => c.copy(expBinIBytes = true)).
- text("use exponential binning strategy for flow byte counts")
-
- opt[Boolean]("expbinpkts").optional().
- action((x, c) => c.copy(expBinIPkts = true)).
- text("use exponential binning strategy for flow packet counts")
-
- opt[Boolean]("protocol").optional().
- action((x, c) => c.copy(useProtocol = true)).
- text("incorporate flow protocol into netflow word creation")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala b/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
deleted file mode 100644
index 49471cc..0000000
--- a/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
+++ /dev/null
@@ -1,90 +0,0 @@
-package org.apache.spot.natetest
-
-
-import org.apache.spot.netflow._
-import org.apache.spot.SuspiciousConnectsArgumentParser._
-import org.apache.log4j.{Level, LogManager, Logger}
-import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
-import org.apache.spark.sql.types.{LongType, StructField, StructType}
-import org.apache.spark.sql.{Row, SQLContext}
-import org.apache.spark.{SparkConf, SparkContext}
-/**
- * Top level entrypoint to execute suspicious connections analysis on network data.
- * Supported analyses:
- * flow : netflow data
- * dns : DNS server logs
- * proxy : proxy server logs
- */
-
-object FlowTune {
-
- /**
- * Execute suspicious connections analysis on network data.
- *
- * @param args Command line arguments.
- */
- def main(args: Array[String]) {
-
- val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
-
- Logger.getLogger("org").setLevel(Level.WARN)
- Logger.getLogger("akka").setLevel(Level.OFF)
-
- val sparkConfig = new SparkConf().setAppName("Netflow tuning")
- val sparkContext = new SparkContext(sparkConfig)
- val sqlContext = new SQLContext(sparkContext)
-
- val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
- val FeedbackFile = ""
- val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11.baseline.scored.pqt"
-
-
- val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
-
- val config = SuspiciousConnectsConfig(analysis = "flow",
- inputPath = InputParquetFile,
- feedbackFile = FeedbackFile,
- duplicationFactor = 1000,
- topicCount = 20,
- userDomain = "intel",
- hdfsScoredConnect = ScoredParquetFile,
- threshold = 1.0d,
- maxResults = -1,
- outputDelimiter = "\t",
- ldaPRGSeed = Some(0xdeadbeef),
- ldaMaxiterations = 20,
- ldaAlpha = 1.02,
- ldaBeta = 1.001)
-
-
- val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sparkContext, sqlContext, logger, inputDataFrame).cache()
-
- println("let me show you ten: ")
- scoredDF.show(10)
-
- // let's do the ROC first
-
- val E = new BinaryClassificationMetrics(scoredDF.select("score", "is_attack").rdd.map({case (r: Row) => (r.getAs[Double](0), r.getAs[Double](1))}))
-
- val auc = E.areaUnderROC()
-
- println("Here's your area under the ROC curve: " + auc)
-
- System.exit(0)
-
- println("and now we have a dataframe with all of the ranked attacks")
-
- val newSchema = StructType(scoredDF.schema.fields ++ Array(StructField("rank", LongType, false)))
-
- val newRowsRDD = scoredDF.rdd.zipWithIndex().map({case (r,i) => Row.fromSeq(r.toSeq ++ Array[Any](i))})
-
- val indexedAll = sqlContext.createDataFrame(newRowsRDD, newSchema)
-
- val indexedAttacks = indexedAll.filter(indexedAll("is_attack") === 1)
-
- indexedAttacks.show(20)
- // i guess I will need to save this somewhere
- }
-
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
index 4ff716e..f5e1a4b 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
@@ -35,18 +35,8 @@ case class FlowWords(srcWord: String, dstWord: String)
/**
* Contains methods and Spark SQL udf objects for calculation of netflow words from netflow records.
*
- * @param timeCuts Quantile cut-offs for the time of day. Time of day is a floating point number
- * >= 0.0 and < 24.0
- * @param ibytCuts Quantile cut-offs for the inbytes.
- * @param ipktCuts Quantile cut-offs if the incoming packet counts.
*/
-class FlowWordCreator(timeCuts: Array[Double],
- ibytCuts: Array[Double],
- ipktCuts: Array[Double],
- useProtocol: Boolean = false,
- hourBinTime: Boolean = false,
- expBinBytes: Boolean = false,
- expBinPackets: Boolean =false ) extends Serializable {
+object FlowWordCreator extends Serializable {
/**
@@ -102,33 +92,20 @@ class FlowWordCreator(timeCuts: Array[Double],
Try {
val timeOfDay: Double = hour.toDouble + minute.toDouble / 60 + second.toDouble / 3600
- val timeBin = if (hourBinTime == true) {
- hour
- } else {
- Quantiles.bin(timeOfDay, timeCuts)
- }
+ val timeBin = hour
+
val lnOf2 = scala.math.log(2) // natural log of 2
- val ibytBin : Long = if (expBinBytes) {
+ val ibytBin : Long =
scala.math.ceil(scala.math.log(ibyt) / lnOf2).toLong // 0 values should never ever happen
- } else {
- Quantiles.bin(ibyt, ibytCuts)
- }
- val ipktBin : Long = if (expBinPackets) {
- scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
- } else {
- Quantiles.bin(ipkt, ibytCuts)
- }
+ val ipktBin : Long = scala.math.ceil(scala.math.log(ipkt) / lnOf2).toLong // 0 values should never ever happen
val LowToLowPortEncoding = 111111
val HighToHighPortEncoding = 333333
- val proto = if (useProtocol == true) {
- protocol
- } else {
- ""
- }
+ val proto = protocol
+
if (dstPort == 0 && srcPort == 0) {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
index 63152e5..d567f0e 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
@@ -25,28 +25,16 @@ import org.apache.spot.utilities.data.validation.InvalidDataHandler
/**
* Estimate the probabilities of network events using a [[FlowSuspiciousConnectsModel]]
*
- * @param timeCuts Quantile cut-offs for binning time-of-day values when forming words from netflow records.
- * @param ibytCuts Quantile cut-offs for binning ibyt values when forming words from netflow records.
- * @param ipktCuts Quantile cut-offs for binning ipkt values when forming words from netflow records.
* @param topicCount Number of topics used in the topic modelling analysis.
* @param wordToPerTopicProbBC Broadcast map assigning to each word it's per-topic probabilities.
* Ie. Prob [word | t ] for t = 0 to topicCount -1
*/
-class FlowScoreFunction(useProtocol: Boolean,
- hourBinTime : Boolean,
- expBinBytes : Boolean,
- expBinPackets: Boolean,
- timeCuts: Array[Double],
- ibytCuts: Array[Double],
- ipktCuts: Array[Double],
- topicCount: Int,
+class FlowScoreFunction(topicCount: Int,
wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable {
- val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, useProtocol, hourBinTime, expBinBytes, expBinPackets)
-
/**
* Estimate the probability of a netflow connection as distributed from the source IP and from the destination IP
* and assign it the least of these two values.
@@ -78,7 +66,7 @@ class FlowScoreFunction(useProtocol: Boolean,
dstTopicMix: Seq[Double]): Double = {
- val FlowWords(srcWord, dstWord) = flowWordCreator.flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol, ibyt: Long, ipkt: Long)
+ val FlowWords(srcWord, dstWord) = FlowWordCreator.flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, protocol, ibyt: Long, ipkt: Long)
val zeroProb = Array.fill(topicCount) {
0.0
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index 28b0457..e9707cf 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -53,21 +53,11 @@ import scala.util.{Failure, Success, Try}
* @param ipToTopicMix DataFrame assigning a distribution on topics to each document or IP.
* @param wordToPerTopicProb Map assigning to each word it's per-topic probabilities.
* Ie. Prob [word | t ] for t = 0 to topicCount -1
- * @param timeCuts Quantile cut-offs for binning time-of-day values when forming words from netflow records.
- * @param ibytCuts Quantile cut-offs for binning ibyt values when forming words from netflow records.
- * @param ipktCuts Quantile cut-offs for binning ipkt values when forming words from netflow records.
*/
class FlowSuspiciousConnectsModel(topicCount: Int,
ipToTopicMix: DataFrame,
- wordToPerTopicProb: Map[String, Array[Double]],
- useProtocol: Boolean,
- hourBinTime: Boolean,
- expBinBytes: Boolean,
- expBinPackets: Boolean,
- timeCuts: Array[Double],
- ibytCuts: Array[Double],
- ipktCuts: Array[Double]) {
+ wordToPerTopicProb: Map[String, Array[Double]]) {
def score(sc: SparkContext, sqlContext: SQLContext, flowRecords: DataFrame): DataFrame = {
@@ -91,15 +81,8 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
recordsWithIPTopicMixes.selectExpr(schema: _*).withColumnRenamed(TopicProbabilityMix, DstIpTopicMix)
}
- val scoreFunction = new FlowScoreFunction(useProtocol,
- hourBinTime,
- expBinBytes,
- expBinPackets,
- timeCuts,
- ibytCuts,
- ipktCuts,
- topicCount,
- wordToPerTopicProbBC)
+
+ val scoreFunction = new FlowScoreFunction(topicCount, wordToPerTopicProbBC)
val scoringUDF = udf((hour: Int,
@@ -122,8 +105,8 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
srcPort,
dstPort,
protocol,
- ipkt,
ibyt,
+ ipkt,
srcIpTopicMix,
dstIpTopicMix))
@@ -190,85 +173,13 @@ object FlowSuspiciousConnectsModel {
config.feedbackFile,
config.duplicationFactor))
- // create quantile cut-offs
-
- val timeCuts = if (config.flatBinTime == false) {
- logger.info("Using decile based time binning")
- val timeCuts = Quantiles.computeDeciles(totalRecords
- .select(Hour, Minute, Second)
- .rdd
- .flatMap({ case Row(hours: Int, minutes: Int, seconds: Int) =>
- Try {
- (3600 * hours + 60 * minutes + seconds).toDouble
- } match {
- case Failure(_) => Seq()
- case Success(time) => Seq(time)
- }
- }))
-
- logger.info(timeCuts.mkString(","))
- timeCuts
- } else {
- logger.info("Using hour-based time binning")
- null
- }
-
-
-
- val ibytCuts = if (config.expBinIBytes == false) {
- logger.info("Using decile based ibytes binning")
- val ibytCuts = Quantiles.computeDeciles(totalRecords
- .select(Ibyt)
- .rdd
- .flatMap({ case Row(ibyt: Long) =>
- Try {
- ibyt.toDouble
- } match {
- case Failure(_) => Seq()
- case Success(ibyt) => Seq(ibyt)
- }
- }))
-
- logger.info(ibytCuts.mkString(","))
- ibytCuts
- } else {
- logger.info("using exponential binning strategy for flow ibytes")
- null
- }
-
-
-
- logger.info("calculating pkt cuts")
-
- val ipktCuts = if (config.expBinIPkts == false) {
- logger.info("Using quintile based ipkt count binning")
- val ipktCuts =
- Quantiles.computeQuintiles(totalRecords
- .select(Ipkt)
- .rdd
- .flatMap({ case Row(ipkt: Long) =>
- Try {
- ipkt.toDouble
- } match {
- case Failure(_) => Seq()
- case Success(ipkt) => Seq(ipkt)
- }
- }))
-
- logger.info(ipktCuts.mkString(","))
- ipktCuts
- } else {
- logger.info("Using exponential binning for ipkts")
- null
- }
- // simplify dflow log entries into "words"
- val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, config.useProtocol, config.flatBinTime, config.expBinIBytes, config.expBinIPkts)
+ // simplify netflow log entries into "words"
- val dataWithWords = totalRecords.withColumn(SourceWord, flowWordCreator.srcWordUDF(ModelColumns: _*))
- .withColumn(DestinationWord, flowWordCreator.dstWordUDF(ModelColumns: _*))
+ val dataWithWords = totalRecords.withColumn(SourceWord, FlowWordCreator.srcWordUDF(ModelColumns: _*))
+ .withColumn(DestinationWord, FlowWordCreator.dstWordUDF(ModelColumns: _*))
dataWithWords.cache()
dataWithWords.count
@@ -312,13 +223,6 @@ object FlowSuspiciousConnectsModel {
new FlowSuspiciousConnectsModel(config.topicCount,
ipToTopicMix,
- wordToPerTopicProb,
- config.useProtocol,
- config.flatBinTime,
- config.expBinIBytes,
- config.expBinIPkts,
- timeCuts,
- ibytCuts,
- ipktCuts)
+ wordToPerTopicProb)
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index c16fb13..afded0a 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -89,6 +89,67 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
+
+
+
+ "netflow suspicious connects" should "correctly identify time-of-day anomalies with testing config" in {
+
+ val testConfig2 = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = "",
+ feedbackFile = "",
+ duplicationFactor = 1,
+ topicCount = 20,
+ hdfsScoredConnect = "",
+ threshold = 1.0d,
+ maxResults = 1000,
+ outputDelimiter = "\t",
+ ldaPRGSeed = None,
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+ val logger = LogManager.getLogger("SuspiciousConnectsAnalysis")
+ logger.setLevel(Level.INFO)
+
+ val anomalousRecord = FlowRecord("2016-05-05 00:11:01", 2016, 5, 5, 0, 0, 1, 0.972f, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39, 12522, 0, 0)
+ val typicalRecord = FlowRecord("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972f, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39, 12522, 0, 0)
+
+
+ val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord,
+ typicalRecord, typicalRecord, typicalRecord, typicalRecord))
+
+
+
+ val flows : DataFrame = FlowSuspiciousConnectsModel.cleanData(data)
+
+
+ logger.info("Fitting probabilistic model to data")
+ val model =
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, testConfig2, flows)
+
+ logger.info("Identifying outliers")
+ val scoredData = model.score(sparkContext, sqlContext, flows)
+
+ val anomalyScore = scoredData.filter(scoredData(Hour) === 0).first().getAs[Double](Score)
+ val typicalScores = scoredData.filter(scoredData(Hour) === 13).collect().map(_.getAs[Double](Score))
+
+ Math.abs(anomalyScore - 0.1d) should be < 0.01
+ typicalScores.length shouldBe 9
+ Math.abs(typicalScores(0) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(1) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(2) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(3) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(4) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(5) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(6) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(7) - 0.9d) should be < 0.01
+ Math.abs(typicalScores(8) - 0.9d) should be < 0.01
+
+
+ }
+
+
+
"filterAndSelectInvalidFlowRecords" should "return invalid records" in {
val invalidFlowRecords = FlowSuspiciousConnectsAnalysis
@@ -189,4 +250,5 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
val scoredFlowRecordsDF = sqlContext.createDataFrame(scoredFlowRecordsRDD, scoredFlowRecordsSchema)
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/78b0090a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
index a3d8a0a..c3df375 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowWordCreatorTest.scala
@@ -30,20 +30,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val second = 32
val protocol = "UDP"
- val ibyts = 222L
- val ipkts = 3L
-
- val timeCuts = Array(2.4, 4.8, 7.2, 9.6, 12.0, 14.4, 16.8, 19.2, 21.6, 24.0)
- val ipktCuts = Array(10d, 20d, 30d, 40d, 50d, 60d, 70d, 80d, 90d, 100d)
- val ibytCuts = Array(100d, 200d, 300d, 400d, 500d)
-
- val expectedIpktBin = 0
- val expectedIbytBin = 2
- val expectedTimeBin = 5
-
-
- val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts)
-
+ val ibyts = 222L // ceil of log2 is 8
+ val ipkts = 3L // ceil of log2 is 2
+
+
// 1. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
"flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1" in {
@@ -51,11 +41,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 23
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_23__5_2_0"
- srcWord shouldBe "23__5_2_0"
+ dstWord shouldBe "-1_23_UDP_12_8_2"
+ srcWord shouldBe "23_UDP_12_8_2"
}
@@ -66,10 +56,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2132
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "23__5_2_0"
- srcWord shouldBe "-1_23__5_2_0"
+ dstWord shouldBe "23_UDP_12_8_2"
+ srcWord shouldBe "-1_23_UDP_12_8_2"
}
// 3. Test when sip is less than dip and sip is not 0 and dport and sport are > 1024 +
@@ -78,10 +68,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 9874
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "333333__5_2_0"
- srcWord shouldBe "333333__5_2_0"
+ dstWord shouldBe "333333_UDP_12_8_2"
+ srcWord shouldBe "333333_UDP_12_8_2"
}
// 4. Test when sip is less than dip and sip is not 0 and dport is 0 but sport is not +
@@ -90,11 +80,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 0
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80__5_2_0"
- srcWord shouldBe "-1_80__5_2_0"
+ dstWord shouldBe "80_UDP_12_8_2"
+ srcWord shouldBe "-1_80_UDP_12_8_2"
}
// 5. Test when sip is less than dip and sip is not 0 and sport is 0 but dport is not +
@@ -105,11 +95,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_43__5_2_0"
- srcWord shouldBe "43__5_2_0"
+ dstWord shouldBe "-1_43_UDP_12_8_2"
+ srcWord shouldBe "43_UDP_12_8_2"
}
// 6. Test when sip is less than dip and sip is not 0 and sport and dport are less or equal than 1024 +
@@ -119,10 +109,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "111111__5_2_0"
- srcWord shouldBe "111111__5_2_0"
+ dstWord shouldBe "111111_UDP_12_8_2"
+ srcWord shouldBe "111111_UDP_12_8_2"
}
// 7. Test when sip is less than dip and sip is not 0 and sport and dport are 0+
@@ -132,10 +122,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "0__5_2_0"
- srcWord shouldBe "0__5_2_0"
+ dstWord shouldBe "0_UDP_12_8_2"
+ srcWord shouldBe "0_UDP_12_8_2"
}
// 8. Test when sip is not less than dip and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0+
@@ -145,10 +135,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_43__5_2_0"
- srcWord shouldBe "43__5_2_0"
+ dstWord shouldBe "-1_43_UDP_12_8_2"
+ srcWord shouldBe "43_UDP_12_8_2"
}
@@ -158,10 +148,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 2435
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80__5_2_0"
- srcWord shouldBe "-1_80__5_2_0"
+ dstWord shouldBe "80_UDP_12_8_2"
+ srcWord shouldBe "-1_80_UDP_12_8_2"
}
@@ -172,10 +162,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "333333__5_2_0"
- srcWord shouldBe "333333__5_2_0"
+ dstWord shouldBe "333333_UDP_12_8_2"
+ srcWord shouldBe "333333_UDP_12_8_2"
}
// 11. Test when sip is not less than dip and dport is 0 but sport is not +
@@ -185,10 +175,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "80__5_2_0"
- srcWord shouldBe "-1_80__5_2_0"
+ dstWord shouldBe "80_UDP_12_8_2"
+ srcWord shouldBe "-1_80_UDP_12_8_2"
}
// 12. Test when sip is not less than dip and sport is 0 but dport is not +
@@ -198,10 +188,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_2435__5_2_0"
- srcWord shouldBe "2435__5_2_0"
+ dstWord shouldBe "-1_2435_UDP_12_8_2"
+ srcWord shouldBe "2435_UDP_12_8_2"
}
// 13. Test when sip is not less than dip and sport and dport are less or equal than 1024
@@ -211,10 +201,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "111111__5_2_0"
- srcWord shouldBe "111111__5_2_0"
+ dstWord shouldBe "111111_UDP_12_8_2"
+ srcWord shouldBe "111111_UDP_12_8_2"
}
// 14. Test when sip is not less than dip and sport and dport are 0
@@ -224,10 +214,10 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val FlowWords(srcWord, dstWord) =
- flowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "0__5_2_0"
- srcWord shouldBe "0__5_2_0"
+ dstWord shouldBe "0_UDP_12_8_2"
+ srcWord shouldBe "0_UDP_12_8_2"
}
// 15. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
@@ -236,13 +226,12 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val dstPort = 23
val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true)
val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_23_TCP_5_2_0"
- srcWord shouldBe "23_TCP_5_2_0"
+ dstWord shouldBe "-1_23_TCP_12_8_2"
+ srcWord shouldBe "23_TCP_12_8_2"
}
@@ -252,12 +241,11 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val srcPort = 0
val dstPort = 0
val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true)
val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "0_TCP_5_2_0"
- srcWord shouldBe "0_TCP_5_2_0"
+ dstWord shouldBe "0_TCP_12_8_2"
+ srcWord shouldBe "0_TCP_12_8_2"
}
@@ -266,9 +254,8 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val srcPort = 2132
val dstPort = 23
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true, true, true, true)
val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "-1_23_UDP_12_8_2"
@@ -282,99 +269,18 @@ class FlowWordCreatorTest extends FlatSpec with Matchers {
val srcPort = 0
val dstPort = 0
val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, true, true, true, true)
val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
+ FlowWordCreator.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
dstWord shouldBe "0_TCP_12_8_2"
srcWord shouldBe "0_TCP_12_8_2"
}
- // 19. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
- "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified time bins" in {
- val srcPort = 2132
- val dstPort = 23
-
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, true, false, false)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_23__12_2_0"
- srcWord shouldBe "23__12_2_0"
- }
-
-
- // 20. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified time bins" in {
- val srcPort = 0
- val dstPort = 0
- val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, true, false, false)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
-
- dstWord shouldBe "0__12_2_0"
- srcWord shouldBe "0__12_2_0"
- }
-
- // 21. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
- "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified byte bins" in {
- val srcPort = 2132
- val dstPort = 23
-
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, true, false)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
- dstWord shouldBe "-1_23__5_8_0"
- srcWord shouldBe "23__5_8_0"
- }
-
-
- // 22. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified byte bins" in {
- val srcPort = 0
- val dstPort = 0
- val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, true,false)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
-
- dstWord shouldBe "0__5_8_0"
- srcWord shouldBe "0__5_8_0"
- }
-
- // 23. Test when sip is less than dip and sip is not 0 and dport is <= 1024 & sport > 1024 and min(dport, sport) !=0 +
- "flowWords" should "create word with ip_pair as sourceIp-destIp, port is dport and dest_word direction is -1 with modified packet bins" in {
- val srcPort = 2132
- val dstPort = 23
-
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, false, true)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
-
-
- dstWord shouldBe "-1_23__5_2_2"
- srcWord shouldBe "23__5_2_2"
-
- }
-
-
- // 24. Test when sip is not less than dip and sport and dport are 0
- it should "create word with ip_pair as destIp-sourceIp, port is max(0,0) and both words direction is 1 (not showing) with modified packet bins" in {
- val srcPort = 0
- val dstPort = 0
- val protocol = "TCP"
- val flowWordCreatorProtocol = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts, false, false, false, true)
- val FlowWords(srcWord, dstWord) =
- flowWordCreatorProtocol.flowWords(hour, minute, second, srcPort, dstPort, protocol, ibyts, ipkts)
-
- dstWord shouldBe "0__5_2_2"
- srcWord shouldBe "0__5_2_2"
- }
}