Merge pull request apache#32 from wangtuanjie/wtj

x

Author: wangtuanjie

scala/base/src/main/scala/com/pili/Json.scala

@@ -21,7 +21,7 @@ object Json {
 
   def marshalAsString(obj: AnyRef): String = mapper.writeValueAsString(obj)
 
-  def unmarsha[T: ClassTag](value: Array[Byte]): T = mapper.readValue(value, getClassTag[T].runtimeClass.asInstanceOf[Class[T]])
+  def unmarshal[T: ClassTag](value: Array[Byte]): T = mapper.readValue(value, getClassTag[T].runtimeClass.asInstanceOf[Class[T]])
 
   def unmarshal[T: ClassTag](in: InputStream): T = mapper.readValue(in, getClassTag[T].runtimeClass.asInstanceOf[Class[T]])
 

scala/base/src/main/scala/com/pili/Mongo.scala

@@ -1,8 +1,11 @@
 package com.pili
 
+import java.util.concurrent.TimeUnit
+
 import scala.collection.JavaConversions._
 
 import com.mongodb.client.MongoCollection
+import com.mongodb.client.model.IndexOptions
 import com.mongodb.{MongoClient, ServerAddress}
 import org.bson.{BsonInt32, Document}
 
@@ -11,9 +14,14 @@ object Mongo {
   def ensureUniqueIndex(coll: MongoCollection[Document], keys: Seq[String]) = {
     val index = new Document()
     keys.foreach(index.append(_, new BsonInt32(1)))
-    val opt = new Document()
-    opt.append("unique", true)
-    coll.createIndex(index)
+    val opt = new IndexOptions().unique(true)
+    coll.createIndex(index, opt)
+  }
+
+  def ensureExpireIndex(coll: MongoCollection[Document], field: String) = {
+    val index = new Document().append(field, new BsonInt32(1))
+    val opt = new IndexOptions().expireAfter(1L, TimeUnit.SECONDS)
+    coll.createIndex(index, opt)
   }
 
   def newMongoClient(a: String): MongoClient = {

scala/biz/src/main/scala/com/pili/biz/QiniuAudit.scala

@@ -0,0 +1,52 @@
+package com.pili.biz
+
+
+object QiniuAudit {
+  def apply(log: String) = {
+    try {
+      Some(new QiniuAudit(log))
+    } catch {
+      case e: Exception => None
+    }
+  }
+}
+
+@SerialVersionUID(311725166982L)
+class QiniuAudit(val log: String) extends java.io.Serializable {
+
+  val fields = log.split("\t")
+  require(fields.length == 12)
+
+  def service: String = fields(1)
+
+  val startTime: Long = fields(2).toLong
+  val endTime: Long = startTime + fields(11).toLong
+
+  def method: String = fields(3)
+
+  def path: String = fields(4)
+
+  def reqBody: String = fields(6)
+
+  val statusCode: Int = fields(7).toInt
+
+  def respBody: String = fields(9)
+
+  val respBodyLen: Long = fields(10).toLong
+
+  override def toString: String = log
+
+  override def hashCode: Int = log.hashCode
+
+  override def equals(a: Any): Boolean = a match {
+    case l: QiniuAudit => log.equals(l.log)
+    case _ => false
+  }
+
+  def elapsed: Long = endTime - startTime
+
+  def contains(key: String): Boolean = fields.exists(_.contains(key))
+
+  def lineContains(key: String): Boolean = log.contains(key)
+
+}

scala/biz/src/main/scala/com/pili/biz/QiniuLog.scala

@@ -1,6 +1,5 @@
 package com.pili.biz
 
-
 // logsync收集回来的日志,会在每行尾添加 \t$sname\t$host
 class QiniuLog(val line: String) {
 

scala/project/build.scala

@@ -12,6 +12,7 @@ object Build extends Build {
       organization := "pili.com",
       libraryDependencies ++= Seq(
         "org.scalatest" %% "scalatest" % "2.2.1" % "test",
+        "org.slf4s" %% "slf4s-api" % "1.7.10",
         "joda-time" % "joda-time" % "2.8",
         "org.mongodb" % "mongo-java-driver" % "3.2.1",
         "com.fasterxml.jackson.core" % "jackson-databind" % "2.4.4",
@@ -26,7 +27,7 @@ object Build extends Build {
     "org.apache.spark" %% "spark-hive" % "1.6.2" % "provided"
   )
 
-  lazy val root = project.in(file(".")).aggregate(base, biz, loadStreamgateAudit)
+  lazy val root = project.in(file(".")).aggregate(base, biz, loadStreamgateAudit, statException)
 
   lazy val base = Project("base", file("base"))
   .settings(defaultSettings: _*)
@@ -41,5 +42,12 @@ object Build extends Build {
   .settings(defaultSettings: _*)
   .settings(libraryDependencies ++= sparkDependencies)
   .settings(libraryDependencies += "org.mongodb.mongo-hadoop" % "mongo-hadoop-core" % "1.5.2" % "provided")
+
+  lazy val statException = Project("statException", file("statException"))
+  .dependsOn(base)
+  .dependsOn(biz)
+  .settings(defaultSettings: _*)
+  .settings(libraryDependencies += "org.apache.kafka" %% "kafka" % "0.8.2.1")
+  .settings(libraryDependencies += "org.apache.flume" % "flume-ng-sdk" % "1.5.2")
 }
 

scala/statException/src/main/scala/com/pili/statexception/Forward.scala

@@ -0,0 +1,148 @@
+package com.pili.statexception
+
+import java.util.Date
+
+import com.mongodb.MongoClient
+import com.mongodb.client.MongoCollection
+import com.mongodb.client.model.UpdateOptions
+import org.bson.Document
+import org.joda.time.format.DateTimeFormat
+
+import com.pili.{Json, Mongo, Props}
+
+import com.pili.biz.QiniuLog
+
+
+object Forward {
+
+  case class Entry(date: Date, hub: String, stream: String, reason: String, remote: String, local: String)
+
+
+  case class ConnectInfo(
+                          Role: String,
+                          URL: String,
+                          RemoteAddr: String,
+                          LocalAddr: String
+                        )
+
+  case class OnStopInfo(streamID: String, infos: Seq[ConnectInfo], detail: String)
+
+  val dateFmt = DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss.SSSSSS")
+
+  // scalastyle:off
+  // 2016/08/03 00:36:24.694523 [ZUwqAAEbL5_lCmcU][INFO] qiniu.com/pili.v1/streamgate/callback.go:319: onStop {"streamID":"z1.1024.1024","infos":[{"IsSource":false,"Role":"dialer","URL":"rtmp://wtj.com/1024/1024?role=customForward","Method":"publish","ConnID":"2rT6m841Si6IpVAL","RemoteAddr":"172.21.1.102:1935","LocalAddr":"10.200.20.28:55600","Source":"","Status":"disconnected","Length":64809,"Frames":{"Video":14,"Audio":26,"Data":0},"Delay":0,"Duration":0,"ExtraData":{"SourceID":"4gQIPJTcScqDKrEL","Dialer":{"url":"rtmp://wtj.com/1024/1024?role=customForward","method":"publish","timeoutSecond":30},"PingInterval":0,"CallbackData":{"hub":"1024","reqId":"Y3M4LgAAAAAb0BxM","streamId":"z1.1024.1024","uid":"1380518997","url":"rtmp://wtj.com/1024/1024?role=customForward"}},"NoGopCache":false}],"detail":"EOF"}
+  // scalastyle:on
+  def parse(line: String): Option[Entry] = {
+    if (line.indexOf(": onStop") == -1) {
+      return None
+    }
+    if (line.indexOf("base.ExtraData") != -1) {
+      // 过滤未升级前 来的日志格式
+      return None
+    }
+    val d = dateFmt.parseDateTime(line.substring(0, 26)).toDate
+    val l = line.indexOf("{")
+    val r = line.lastIndexOf("}")
+    val jl = line.substring(l, r + 1)
+    val info = Json.unmarshal[OnStopInfo](jl)
+
+    for (x <- info.infos) {
+      if (x.URL.indexOf("role=customForward") > 0) {
+        val fields = info.streamID.split("\\.")
+        val hub = fields(1)
+        val stream = fields(2)
+        val remote = x.RemoteAddr.split(":")(0)
+        val local = x.LocalAddr.split(":")(0)
+        return Some(new Entry(d, hub, stream, info.detail, remote, local))
+      }
+    }
+    return None
+  }
+}
+
+class Forward(val conf: Map[String, String]) extends Runnable {
+
+  import Forward._
+
+
+  val kafkaConf = Props.subProps(conf, "kafka")
+  val expireDay = conf("expire.day").toInt
+  val expireMs = 24 * 3600 * 1000 * expireDay
+
+  val mgoAddr = conf("mongo.addr")
+  val mgoDB = conf("mongo.db")
+  val mgoColl = conf("mongo.coll.forward")
+
+
+  def run(): Unit = {
+    val mcli: MongoClient = Mongo.newMongoClient(mgoAddr)
+    val coll = mcli.getDatabase(mgoDB).getCollection(mgoColl)
+    Mongo.ensureUniqueIndex(coll, Seq("time", "hub", "key"))
+    Mongo.ensureExpireIndex(coll, "expireAt")
+
+
+    var n = 0L
+    val msgStream = Util.consumerTopic(kafkaConf("topic.forward"), kafkaConf)
+
+    msgStream.foreach(msgs => {
+      msgs.foreach(msg => {
+        try {
+          n += 1
+          if (n % 10000 == 0) {
+            println(s"forward handled $n")
+          }
+
+          val line = Util.flumeEventDecode(msg.message())
+          if (line.isDefined) {
+            val l = new QiniuLog(line.get)
+            val entry = parse(l.log)
+            if (entry.isDefined) {
+              val reason = reasonOf(entry.get.reason)
+              if (reason.isDefined) {
+                insert2Mgo(coll, entry.get)
+              }
+            }
+          }
+        } catch {
+          case e: Exception => {
+            e.printStackTrace()
+          }
+        }
+      })
+    })
+  }
+
+
+  def insert2Mgo(coll: MongoCollection[Document], e: Entry) = {
+    val query = new Document("time", e.date)
+      .append("hub", e.hub)
+      .append("key", e.stream)
+
+    val opt = new UpdateOptions().upsert(true)
+
+    val update = new Document("$set", new Document()
+      .append("remote", e.remote)
+      .append("local", e.local)
+      .append("reason", e.reason)
+      .append("expireAt", new Date(e.date.getTime + expireMs)))
+
+    coll.updateOne(query, update, opt)
+  }
+
+
+  def reasonOf(reason: String): Option[String] = {
+    reason match {
+      case "Connection Timeout" => Some("connection timeout") // 数据超时
+      case "EOF" => None
+      case "unexpected EOF" => None
+      case "source closed" => None
+      case _ => {
+        (reason.indexOf("connection reset by peer") > 0) match {
+          case true => Some("connection reset by peer")
+          case _ => Some("unknown")
+        }
+      }
+    }
+  }
+
+}

scala/statException/src/main/scala/com/pili/statexception/Main.scala

@@ -0,0 +1,20 @@
+package com.pili.statexception
+
+import com.pili.Props
+
+object Main extends App {
+
+  if (args.length != 1) {
+    System.err.println("usage: ./bin <config file>")
+    System.exit(1)
+  }
+
+  val conf = Props.load(args(0))
+  val publisher = new Thread(new Publish(conf))
+  val forwarder = new Thread(new Forward(conf))
+  publisher.start()
+  forwarder.start()
+
+  publisher.join()
+  forwarder.join()
+}