Sending ScalikeJDBC query logs to Fluentd

application.conf

db.default.driver="org.postgresql.Driver"
db.default.url="jdbc:postgresql://localhost:5432/scalikejdbc"
db.default.user="sa"
db.default.password="sa"

build.sbt

scalaVersion := "2.10.2"

libraryDependencies ++= Seq(
  "com.github.seratch"  %% "scalikejdbc"               % "1.6.8",
  "com.github.seratch"  %% "scalikejdbc-interpolation" % "1.6.8",
  "com.github.seratch"  %% "scalikejdbc-config"        % "1.6.8",
  "com.github.seratch"  %% "scalikejdbc-async"         % "0.2.5",
  "com.github.mauricio" %% "postgresql-async"          % "0.2.6",
  "com.github.mauricio" %% "mysql-async"               % "0.2.6",
  "org.postgresql"      %  "postgresql"                % "9.2-1003-jdbc4",
  "mysql"               %  "mysql-connector-java"      % "5.1.26",
  "org.fluentd"         %% "fluent-logger-scala"       % "0.4.0",
  "org.scalatest"       %% "scalatest"                 % "1.9.1"            % "test"
)

fluent.conf

# fluentd -c fluent.conf -vv
<source>
  type forward
</source>

<match scalikejdbc.**>
  type stdout
</match>

SampleSpec.scala

package example

import scalikejdbc._, SQLInterpolation._, config._, async._
import org.fluentd.logger.scala._
import org.scalatest._, matchers._
import scala.concurrent._, duration.DurationInt, ExecutionContext.Implicits.global

class SampleSpec extends FlatSpec with ShouldMatchers {

  case class Name(id: Long, kanji: String, kana: String)
  object Name extends SQLSyntaxSupport[Name] {
    override val tableName = "names"
    override val columnNames = Seq("id", "kanji", "kana")
    def apply(n: ResultName[Name])(rs: WrappedResultSet): Name = new Name(
      id = rs.long(n.id),
      kanji = rs.string(n.kanji),
      kana = rs.string(n.kana)
    )
  }

  val logger = FluentLoggerFactory.getLogger("scalikejdbc")
  val n = Name.syntax("n")

  it should "work" in {
    DBs.setupAll()

    DB autoCommit { implicit s =>
       try sql"drop table names;".execute.apply()
       catch { case e: Exception => }

       sql"""
         create table names (
           id bigserial primary key not null,
           kanji varchar(64) not null,
           kana varchar(64) not null);
       """.execute.apply()

       val c = Name.column
       withSQL { insert.into(Name).namedValues(c.kanji -> "瀬良和弘", c.kana -> "せらかずひろ") }.update.apply()
       withSQL { insert.into(Name).namedValues(c.kanji -> "瀬良勇人", c.kana -> "せらゆうと") }.update.apply()
    }

    GlobalSettings.queryCompletionListener = (sql: String, params: Seq[Any], millis: Long) => {
      logger.log("completion", Map(
        "sql" -> sql,
        "params" -> params.mkString("[", ",", "]"),
        "millis" -> millis))
    }
    GlobalSettings.queryFailureListener = (sql: String, params: Seq[Any], e: Throwable) => {
      logger.log("failure", Map(
        "sql" -> sql,
        "params" -> params.mkString("[", ",", "]"),
        "exception" -> e.getClass.getSimpleName,
        "message" -> e.getMessage))
    }

    DB readOnly { implicit s =>
      val names = withSQL { select.from(Name as n) }.map(Name(n.resultName)).list.apply()
      names.size should equal(2)

      try sql"select xxxx".map(Name(n.resultName)).list.apply()
      catch { case e: Exception => }
    }

    AsyncConnectionPool.singleton("jdbc:postgresql://localhost:5432/scalikejdbc", "sa", "sa")
    val res = AsyncDB withPool { implicit s =>
      sql"select yyyy".map(Name(n.resultName)).single.future()
    }
    try Await.result(res, 5.seconds)
    catch { case e: Exception => }
  }

}

stdout

2013-09-07 16:57:42 +0900 [trace]: plugin/in_forward.rb:152:initialize: accepted fluent socket object_id=70115622766820
2013-09-07 16:57:43 +0900 scalikejdbc.completion: {"sql":"select n.id as i_on_n, n.kanji as k1_on_n, n.kana as k2_on_n from names n","params":"[]","millis":13}
2013-09-07 16:57:43 +0900 scalikejdbc.failure: {"sql":"select xxxx","params":"[]","exception":"PSQLException","message":"ERROR: column \"xxxx\" does not exist\n  Position: 8"}
2013-09-07 16:57:43 +0900 [trace]: plugin/in_forward.rb:193:on_close: closed fluent socket object_id=70115622766820

2013-09-07 17:22:26 +0900 [trace]: plugin/in_forward.rb:152:initialize: accepted fluent socket object_id=70115626756740
2013-09-07 17:22:27 +0900 scalikejdbc.completion: {"sql":"select n.id as i_on_n, n.kanji as k1_on_n, n.kana as k2_on_n from names n","params":"[]","millis":23}
2013-09-07 17:22:27 +0900 scalikejdbc.failure: {"sql":"select xxxx","params":"[]","exception":"PSQLException","message":"ERROR: column \"xxxx\" does not exist\n  Position: 8"}
2013-09-07 17:22:29 +0900 [trace]: plugin/in_forward.rb:193:on_close: closed fluent socket object_id=70115626756740
2013-09-07 17:23:22 +0900 [trace]: plugin/in_forward.rb:152:initialize: accepted fluent socket object_id=70115626793380
2013-09-07 17:23:23 +0900 scalikejdbc.completion: {"sql":"select n.id as i_on_n, n.kanji as k1_on_n, n.kana as k2_on_n from names n","params":"[]","millis":16}
2013-09-07 17:23:23 +0900 scalikejdbc.failure: {"sql":"select xxxx","params":"[]","exception":"PSQLException","message":"ERROR: column \"xxxx\" does not exist\n  Position: 8"}
2013-09-07 17:23:23 +0900 [trace]: plugin/in_forward.rb:193:on_close: closed fluent socket object_id=70115626793380
2013-09-07 17:58:15 +0900 [trace]: plugin/in_forward.rb:152:initialize: accepted fluent socket object_id=70115626952620
2013-09-07 17:58:16 +0900 scalikejdbc.completion: {"sql":"select n.id as i_on_n, n.kanji as k1_on_n, n.kana as k2_on_n from names n","params":"[]","millis":22}
2013-09-07 17:58:17 +0900 scalikejdbc.failure: {"sql":"select xxxx","params":"[]","exception":"PSQLException","message":"ERROR: column \"xxxx\" does not exist\n  Position: 8"}
2013-09-07 17:58:17 +0900 scalikejdbc.failure: {"sql":"select yyyy","params":"[]","exception":"GenericDatabaseException","message":"ErrorMessage(fields=Map(Position -> 8, Line -> 766, File -> parse_expr.c, SQLSTATE -> 42703, Routine -> transformColumnRef, Message -> column \"yyyy\" does not exist, Severity -> ERROR))"}
2013-09-07 17:58:18 +0900 [trace]: plugin/in_forward.rb:193:on_close: closed fluent socket object_id=70115626952620