Lwcapi merge

.travis.yml

@@ -17,3 +17,5 @@ cache:
   directories:
   - $HOME/.sbt
   - $HOME/.ivy2
+services:
+  - redis-server

atlas-lwcapi/src/main/resources/reference.conf

@@ -0,0 +1,38 @@
+atlas {
+  lwcapi {
+    register = {
+      default-frequency = 60000
+    }
+
+    redis = {
+      host = "localhost"
+      port = 6379
+      ttl = 300000 // in milliseconds, default 5 minutes
+      prefix = "" // default empty (could be omitted)
+    }
+  }
+
+  akka {
+    actors = ${?atlas.akka.actors} [
+      {
+        name = "lwc.expressiondb"
+        class = "com.netflix.atlas.lwcapi.ExpressionDatabaseActor"
+      },
+      {
+        name = "lwc.subscribe"
+        class = "com.netflix.atlas.lwcapi.SubscribeActor"
+      },
+      {
+        name = "lwc.evaluate"
+        class = "com.netflix.atlas.lwcapi.EvaluateActor"
+      }
+    ]
+
+    api-endpoints = ${?atlas.akka.api-endpoints} [
+      "com.netflix.atlas.lwcapi.EvaluateApi",
+      "com.netflix.atlas.lwcapi.ExpressionApi",
+      "com.netflix.atlas.lwcapi.SubscribeApi",
+      "com.netflix.atlas.lwcapi.StreamApi"
+    ]
+  }
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/ApiSettings.scala

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import com.netflix.atlas.config.ConfigManager
+import com.typesafe.config.Config
+
+object ApiSettings extends ApiSettings(ConfigManager.current)
+
+class ApiSettings(root: => Config) {
+  private def config = root.getConfig("atlas.lwcapi")
+
+  def defaultFrequency: Int = config.getInt("register.default-frequency")
+
+  def redisHost: String = config.getString("redis.host")
+  def redisPort: Int = config.getInt("redis.port")
+  def redisTTL: Int = config.getInt("redis.ttl")
+  def redisPrefix: String = {
+    if (config.hasPath("redis.prefix")) {
+      if (config.getString("redis.prefix").nonEmpty)
+        config.getString("redis.prefix") + "."
+      else
+        ""
+    } else ""
+  }
+
+  def redisPrefixFor(suffix: String): String = {
+    s"$redisPrefix$suffix"
+  }
+
+  def stripRedisPrefix(s: String): String = {
+    if (redisPrefix.isEmpty) s else s.stripPrefix(redisPrefix)
+  }
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/DatabaseService.scala

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import javax.inject.Singleton
+
+import com.netflix.iep.service.{AbstractService, State}
+import com.typesafe.scalalogging.StrictLogging
+
+@Singleton
+class DatabaseService extends AbstractService with StrictLogging {
+  @volatile private var started = false
+
+  def setState(value: Boolean): Unit = started = value
+
+  override def isHealthy: Boolean = state == State.RUNNING && started
+
+  override def startImpl(): Unit = {
+    logger.info("Starting Database service monitor")
+  }
+
+  override def stopImpl(): Unit = {
+    logger.info("Stopping Database service monitor")
+  }
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/EvaluateActor.scala

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import javax.inject.Inject
+
+import akka.actor.{Actor, ActorLogging}
+import com.netflix.atlas.akka.DiagnosticMessage
+import com.netflix.atlas.lwcapi.StreamApi._
+import com.netflix.spectator.api.Registry
+import spray.http.{HttpResponse, StatusCodes}
+
+class EvaluateActor @Inject() (registry: Registry, sm: SubscriptionManager) extends Actor with ActorLogging {
+  import com.netflix.atlas.lwcapi.EvaluateApi._
+
+  private val evalsId = registry.createId("atlas.lwcapi.evaluate.count")
+  private val itemsId = registry.createId("atlas.lwcapi.evaluate.itemCount")
+  private val actorsId = registry.createId("atlas.lwcapi.evaluate.actorCount")
+  private val uninterestingId = registry.createId("atlas.lwcapi.evaluate.uninterestingCount")
+
+  def receive = {
+    case EvaluateRequest(_, Nil) =>
+      DiagnosticMessage.sendError(sender(), StatusCodes.BadRequest, "empty expression payload")
+    case EvaluateRequest(timestamp, items) =>
+      evaluate(timestamp, items)
+      sender() ! HttpResponse(StatusCodes.OK)
+    case _ =>
+      DiagnosticMessage.sendError(sender(), StatusCodes.BadRequest, "unknown payload")
+  }
+
+  private def evaluate(timestamp: Long, items: List[Item]): Unit = {
+    registry.counter(evalsId).increment()
+    registry.counter(itemsId).increment(items.size)
+    items.foreach { item =>
+      val actors = sm.actorsForExpression(item.id)
+      if (actors.nonEmpty) {
+        registry.counter(actorsId).increment(actors.size)
+        val message = SSEMetric(timestamp, item)
+        actors.foreach(actor => actor ! message)
+      } else {
+        registry.counter(uninterestingId).increment()
+      }
+    }
+  }
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/EvaluateApi.scala

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import akka.actor.ActorRefFactory
+import com.netflix.atlas.akka.WebApi
+import com.netflix.atlas.json.{Json, JsonSupport}
+import com.netflix.atlas.lwcapi.StreamApi.SSEMessage
+import spray.routing.RequestContext
+
+class EvaluateApi(implicit val actorRefFactory: ActorRefFactory) extends WebApi {
+  import EvaluateApi._
+
+  private val evaluateRef = actorRefFactory.actorSelection("/user/lwc.evaluate")
+
+  def routes: RequestContext => Unit = {
+    path("lwc" / "api" / "v1" / "evaluate") {
+       post { ctx => handleReq(ctx) }
+    }
+  }
+
+  private def handleReq(ctx: RequestContext): Unit = {
+    val request = EvaluateRequest.fromJson(ctx.request.entity.asString)
+    evaluateRef.tell(request, ctx.responder)
+  }
+}
+
+object EvaluateApi {
+  type TagMap = Map[String, String]
+
+  case class Item(id: String, tags: TagMap, value: Double) extends JsonSupport
+
+  case class EvaluateRequest(timestamp: Long, metrics: List[Item]) extends JsonSupport
+  object EvaluateRequest {
+    def fromJson(json: String) = Json.decode[EvaluateRequest](json)
+  }
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/ExpressionApi.scala

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import javax.inject.Inject
+
+import akka.actor.ActorRefFactory
+import com.netflix.atlas.akka.WebApi
+import com.netflix.atlas.json.{Json, JsonSupport}
+import com.netflix.spectator.api.Registry
+import com.typesafe.scalalogging.StrictLogging
+import spray.http.{HttpResponse, StatusCodes}
+import spray.routing.RequestContext
+
+case class ExpressionApi @Inject()(alertmap: ExpressionDatabase,
+                                   registry: Registry,
+                                   implicit val actorRefFactory: ActorRefFactory)
+  extends WebApi with StrictLogging {
+  import ExpressionApi._
+
+  private val defaultURL = "http://..."
+
+  private val expressionFetchesId = registry.createId("atlas.lwcapi.expressions.fetches")
+  private val expressionCount = registry.distributionSummary("atlas.lwcapi.expressions.count")
+
+  def routes: RequestContext => Unit = {
+    path("lwc" / "api" / "v1" / "expressions" / Segment) { (cluster) =>
+      get { ctx => handleReq(ctx, cluster) }
+    }
+  }
+
+  private def handleReq(ctx: RequestContext, cluster: String): Unit = {
+    val expressions = alertmap.expressionsForCluster(cluster)
+    val json = Return(expressions).toJson
+    ctx.responder ! HttpResponse(StatusCodes.OK, entity = json)
+    registry.counter(expressionFetchesId.withTag("cluster", cluster)).increment()
+    expressionCount.record(expressions.size)
+  }
+}
+
+object ExpressionApi {
+  case class Return(expressions: List[ExpressionWithFrequency]) extends JsonSupport
+}

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/ExpressionDatabase.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2014-2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.atlas.lwcapi
+
+import com.netflix.atlas.core.model.Query
+import com.netflix.atlas.lwcapi.ExpressionSplitter.SplitResult
+
+abstract class ExpressionDatabase {
+  def addExpr(expr: ExpressionWithFrequency, queries: Query): Boolean
+  def delExpr(id: String): Boolean
+  def hasExpr(id: String): Boolean
+  def expr(id: String): Option[ExpressionWithFrequency]
+  def expressionsForCluster(cluster: String): List[ExpressionWithFrequency]
+}