Joal has submitted this change and it was merged.
Change subject: Report RESTBase traffic metrics to Graphite
......................................................................
Report RESTBase traffic metrics to Graphite
This Spark Job aims to Run hourly and report restbase request counts
to Graphite. It will be scheduled via Oozie.
Bug: T109547
Change-Id: I1dd47de9aaa8f80df9a1db1db8c375d07f5ca950
---
M pom.xml
M refinery-core/pom.xml
A
refinery-core/src/main/scala/org/wikimedia/analytics/refinery/core/GraphiteClient.scala
M refinery-job/pom.xml
A
refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RESTBaseMetrics.scala
5 files changed, 272 insertions(+), 14 deletions(-)
Approvals:
Joal: Verified; Looks good to me, approved
diff --git a/pom.xml b/pom.xml
index 75fc600..6174520 100644
--- a/pom.xml
+++ b/pom.xml
@@ -164,11 +164,25 @@
</dependency>
<dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_2.10</artifactId>
<version>2.2.4</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>com.github.nscala-time</groupId>
+ <artifactId>nscala-time_2.10</artifactId>
+ <version>2.0.0</version>
+ </dependency>
+
</dependencies>
</dependencyManagement>
diff --git a/refinery-core/pom.xml b/refinery-core/pom.xml
index 33e0615..063cef7 100644
--- a/refinery-core/pom.xml
+++ b/refinery-core/pom.xml
@@ -74,10 +74,61 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_2.10</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.github.nscala-time</groupId>
+ <artifactId>nscala-time_2.10</artifactId>
+ </dependency>
+
</dependencies>
<build>
<plugins>
+ <plugin>
+ <groupId>org.scala-tools</groupId>
+ <artifactId>maven-scala-plugin</artifactId>
+ <version>2.15.2</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest-maven-plugin</artifactId>
+ <version>1.0</version>
+ <configuration>
+
<reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <filereports>WDF TestSuite.txt</filereports>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
@@ -97,6 +148,7 @@
</execution>
</executions>
</plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
@@ -108,6 +160,7 @@
</systemPropertyVariables>
</configuration>
</plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
diff --git
a/refinery-core/src/main/scala/org/wikimedia/analytics/refinery/core/GraphiteClient.scala
b/refinery-core/src/main/scala/org/wikimedia/analytics/refinery/core/GraphiteClient.scala
new file mode 100644
index 0000000..4b40835
--- /dev/null
+++
b/refinery-core/src/main/scala/org/wikimedia/analytics/refinery/core/GraphiteClient.scala
@@ -0,0 +1,86 @@
+package org.wikimedia.analytics.refinery.core
+
+import java.io.OutputStream
+import java.net.Socket
+
+import org.joda.time.DateTimeUtils
+
+/**
+ * Simple GraphiteClient in Scala
+ * Creates a Socket and writes to it,
+ * based on the plaintext protocol supported by Carbon
+ *
+ * See:
http://graphite.readthedocs.org/en/latest/feeding-carbon.html#the-plaintext-protocol
+ * for details
+ *
+ * Supports sendOnce to open a connection, send a message, close connection.
+ * To write multiple messages at a time, something like
+ *
+ * val graphite = GraphiteClient('localhost')
+ * val conn = graphite.connection()
+ * m1 = graphite.message('foo.bar', 20)
+ * m2 = graphite.message('foo.baz', 30)
+ * conn.write(m1)
+ * conn.write(m2)
+ * conn.close()
+ *
+ */
+class GraphiteClient(host:String, port: Int = 2003) {
+
+ /**
+ * Connection class that wraps the Socket creation,
+ * and writing to the socket functionality, so the GraphiteClient
+ * can be extended easily to support sending multiple messages, etc
+ * @param host Graphite host url
+ * @param port Graphite port
+ */
+ class Connection(host:String, port:Int) {
+ val socket:Socket = new Socket(host, port)
+ val out:OutputStream = socket.getOutputStream
+
+ def write(data:String) = {
+ out.write(data.getBytes())
+ out.flush()
+ }
+
+ def close() = {
+ out.close
+ socket.close
+ }
+ }
+
+ /**
+ * Create an instance of Connection
+ *
+ * @return Instance of Connection that wraps the socket
+ * creation and writing to socket functionality
+ */
+ def connection() = {
+ new Connection(host, port)
+ }
+
+ /**
+ * Helper to create a message string following Carbon's plaintext protocol
+ *
+ * @param metric Name of the graphite metric, e.g foo.bar
+ * @param value Value of the metric
+ * @param timestamp Timestamp in seconds, defaults to current time
+ * @return Formatted string for plaintext protocol
+ */
+ def message(metric:String, value:Long, timestamp:Long =
DateTimeUtils.currentTimeMillis() / 1000) = {
+ "%s %d %d\n".format(metric, value, timestamp)
+ }
+
+ /**
+ * Helper that opens a connection, sends a message, and closes connection
+ * @param metric Name of the graphite metric, e.g foo.bar
+ * @param value Value of the metric
+ * @param timestamp Timestamp in seconds
+ */
+ def sendOnce(metric:String, value:Long, timestamp:Long) = {
+ val conn = connection()
+ conn.write(message(metric, value, timestamp))
+ conn.close()
+ }
+
+}
diff --git a/refinery-job/pom.xml b/refinery-job/pom.xml
index a30ffdc..57cf3f6 100644
--- a/refinery-job/pom.xml
+++ b/refinery-job/pom.xml
@@ -21,13 +21,6 @@
</dependency>
<dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>${scala.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>${spark.version}</version>
@@ -55,12 +48,6 @@
</dependency>
<dependency>
- <groupId>org.scalatest</groupId>
- <artifactId>scalatest_2.10</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
@@ -73,9 +60,20 @@
</dependency>
<dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_2.10</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>com.github.nscala-time</groupId>
<artifactId>nscala-time_2.10</artifactId>
- <version>2.0.0</version>
</dependency>
</dependencies>
@@ -114,6 +112,7 @@
</execution>
</executions>
</plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
diff --git
a/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RESTBaseMetrics.scala
b/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RESTBaseMetrics.scala
new file mode 100644
index 0000000..0bba736
--- /dev/null
+++
b/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RESTBaseMetrics.scala
@@ -0,0 +1,106 @@
+package org.wikimedia.analytics.refinery.job
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.{SparkConf, SparkContext}
+import org.joda.time.DateTime
+import org.wikimedia.analytics.refinery.core.GraphiteClient
+import scopt.OptionParser
+
+/**
+ * Reports metrics for Restbase to graphite
+ *
+ * Usage with spark-submit:
+ * spark-submit \
+ * --class org.wikimedia.analytics.refinery.job.RESTBaseMetrics
+ * /path/to/refinery-job.jar
+ * -y <year> -m <month> -d <day> -h <hour>
+ * [-n <namespace> -w <webrequest-base-path> -g <graphite-host> -p
<graphite-port>]
+ */
+object RESTBaseMetrics {
+
+
+ /**
+ * Config class for CLI argument parser using scopt
+ */
+ case class Params(webrequestBasePath: String =
"hdfs://analytics-hadoop/wmf/data/wmf/webrequest",
+ graphiteHost: String = "localhost",
+ graphitePort: Int = 2003,
+ namespace: String = "restbase.requests",
+ year: Int = 0, month: Int = 0, day: Int = 0, hour: Int = 0)
+
+ /**
+ * Define the command line options parser
+ */
+ val argsParser = new OptionParser[Params]("RESTBase Metrics") {
+ head("RESTBase Metrics", "")
+ note("This job reports RESTBase traffic to graphite hourly")
+ help("help") text ("Prints this usage text")
+
+ opt[String]('w', "webrequest-base-path") optional() valueName ("<path>")
action { (x, p) =>
+ p.copy(webrequestBasePath = if (x.endsWith("/")) x.dropRight(1) else x)
+ } text ("Base path to webrequest data on hadoop. Defaults to
hdfs://analytics-hadoop/wmf/data/wmf/webrequest")
+
+ opt[String]('g', "graphite-host") optional() valueName ("<path>") action {
(x, p) =>
+ p.copy(graphiteHost = x)
+ } text ("Graphite host. Defaults to localhost")
+
+ opt[Int]('p', "graphite-port") optional() valueName ("<path>") action {
(x, p) =>
+ p.copy(graphitePort = x)
+ } text ("Graphite port. Defaults to 2003")
+
+ opt[String]('n', "namespace") optional() valueName ("<path>") action { (x,
p) =>
+ p.copy(namespace = x)
+ } text ("Namespace/prefix for graphite metric. Defaults to
restbase.requests")
+
+ opt[Int]('y', "year") required() action { (x, p) =>
+ p.copy(year = x)
+ } text ("Year as an integer")
+
+ opt[Int]('m', "month") required() action { (x, p) =>
+ p.copy(month = x)
+ } validate { x => if (x > 0 & x <= 12) success else failure("Invalid
month")
+ } text ("Month as an integer")
+
+ opt[Int]('d', "day") required() action { (x, p) =>
+ p.copy(day = x)
+ } validate { x => if (x > 0 & x <= 31) success else failure("Invalid day")
+ } text ("Day of month as an integer")
+
+ opt[Int]('h', "hour") required() action { (x, p) =>
+ p.copy(hour = x)
+ } validate { x => if (x >= 0 & x < 24) success else failure("Invalid hour")
+ } text ("Hour of day as an integer (0-23)")
+
+ }
+
+ def countRESTBaseURIs(parquetData: DataFrame): Long = {
+ parquetData.filter("uri_path like '%/api/rest_v1%'").count
+ }
+
+ def main(args: Array[String]): Unit = {
+ argsParser.parse(args, Params()) match {
+ case Some(params) => {
+ // Initial Spark setup
+ val conf = new SparkConf().setAppName("RESTBaseMetrics")
+ val sc = new SparkContext(conf)
+ val sqlContext = new SQLContext(sc)
+ sqlContext.setConf("spark.sql.parquet.compression.codec", "snappy")
+
+ // Define the path to load data in Parquet format
+ val parquetDataPath =
"%s/webrequest_source=text/year=%d/month=%d/day=%d/hour=%d"
+ .format(params.webrequestBasePath, params.year, params.month,
params.day, params.hour)
+
+ // Define time, metric, Compute request count
+ val time = new DateTime(params.year, params.month, params.day,
params.hour, 0)
+ val metric = "%s.varnish_requests".format(params.namespace)
+ val requestCount =
countRESTBaseURIs(sqlContext.parquetFile(parquetDataPath))
+
+ // Send to graphite
+ val graphite = new GraphiteClient(params.graphiteHost,
params.graphitePort)
+ graphite.sendOnce(metric, requestCount, time.getMillis / 1000)
+ }
+ case None => sys.exit(1)
+ }
+ }
+
+}
--
To view, visit https://gerrit.wikimedia.org/r/234453
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I1dd47de9aaa8f80df9a1db1db8c375d07f5ca950
Gerrit-PatchSet: 7
Gerrit-Project: analytics/refinery/source
Gerrit-Branch: master
Gerrit-Owner: Madhuvishy <[email protected]>
Gerrit-Reviewer: Joal <[email protected]>
Gerrit-Reviewer: Madhuvishy <[email protected]>
Gerrit-Reviewer: Nuria <[email protected]>
Gerrit-Reviewer: Ottomata <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
