| /* |
| * 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.openwhisk.core.monitoring.metrics |
| |
| import akka.http.scaladsl.Http |
| import akka.http.scaladsl.model.{HttpRequest, StatusCodes} |
| import akka.http.scaladsl.unmarshalling.Unmarshal |
| import com.typesafe.config.ConfigFactory |
| import kamon.Kamon |
| import org.junit.runner.RunWith |
| import org.scalatest.junit.JUnitRunner |
| |
| import scala.concurrent.duration._ |
| import scala.util.Try |
| |
| @RunWith(classOf[JUnitRunner]) |
| class OpenWhiskEventsTests extends KafkaSpecBase { |
| behavior of "Server" |
| |
| it should "start working http server" in { |
| val httpPort = freePort() |
| val globalConfig = system.settings.config |
| val config = ConfigFactory.parseString(s""" |
| | akka.kafka.consumer.kafka-clients { |
| | bootstrap.servers = "localhost:$kafkaPort" |
| | } |
| | kamon { |
| | metric { |
| | tick-interval = 50 ms |
| | optimistic-tick-alignment = no |
| | } |
| | } |
| | whisk { |
| | user-events { |
| | port = $httpPort |
| | } |
| | } |
| """.stripMargin).withFallback(globalConfig) |
| |
| val binding = OpenWhiskEvents.start(config).futureValue |
| val res = get("localhost", httpPort, "/ping") |
| res shouldBe Some(StatusCodes.OK, "pong") |
| |
| //Check if metrics using Kamon API gets included in consolidated Prometheus |
| Kamon.counter("fooTest").withoutTags().increment(42) |
| sleep(1.second) |
| val metricRes = get("localhost", httpPort, "/metrics") |
| metricRes.get._2 should include("fooTest") |
| |
| binding.unbind().futureValue |
| } |
| |
| def get(host: String, port: Int, path: String = "/") = { |
| val response = Try { |
| Http() |
| .singleRequest(HttpRequest(uri = s"http://$host:$port$path")) |
| .futureValue |
| }.toOption |
| |
| response.map { res => |
| (res.status, Unmarshal(res).to[String].futureValue) |
| } |
| } |
| } |