Dubbo壓測外掛的實現——基於Gatling
Dubbo 壓測外掛已開源,本文涉及程式碼詳見gatling-dubbo
Gatling 是一個開源的基於 Scala、Akka、Netty 實現的高效能壓測框架,較之其他基於執行緒實現的壓測框架,Gatling 基於 AKKA Actor 模型實現,請求由事件驅動,在系統資源消耗上低於其他壓測框架(如記憶體、連線池等),使得單臺施壓機可以模擬更多的使用者。此外,Gatling 提供了一套簡單高效的 DSL(領域特定語言)方便我們編排業務場景,同時也具備流量控制、壓力控制的能力並提供了良好的壓測報告,所以有贊選擇在 Gatling 基礎上擴充套件分散式能力,開發了自己的全鏈路壓測引擎 MAXIM。全鏈路壓測中我們主要模擬使用者實際使用場景,使用 HTTP 介面作為壓測入口,但有贊目前後端服務中 Dubbo 應用比重越來越高,如果可以知道 Dubbo 應用單機水位將對我們把控系統後端服務能力大有裨益。基於 Gatling 的優勢和在有讚的使用基礎,我們擴充套件 Gatling 開發了 gatling-dubbo 壓測外掛。
外掛主要結構
實現 Dubbo 壓測外掛,需實現以下四部分內容:
- Protocol 和 ProtocolBuild
協議部分,這裡主要定義 Dubbo 客戶端相關內容,如協議、泛化呼叫、服務 URL、註冊中心等內容,ProtocolBuild 則為 DSL 使用 Protocol 的輔助類 - Action 和 ActionBuild
執行部分,這裡的作用是發起 Dubbo 請求,校驗請求結果並記錄日誌以便後續生成壓測報告。ActionBuild 則為 DSL 使用 Action 的輔助類 - Check 和 CheckBuild
檢查部分,全鏈路壓測中我們都使用Json Path
- DSL
Dubbo 外掛的領域特定語言,我們提供了一套簡單易用的 API 方便編寫 Duboo 壓測指令碼,風格上與原生 HTTP DSL 保持一致
Protocol
協議部分由 5 個屬性組成,這些屬性將在 Action 初始化 Dubbo 客戶端時使用,分別是:
- protocol
協議,設定為dubbo
- generic
泛化呼叫設定,Dubbo 壓測外掛使用泛化呼叫發起請求,所以這裡設定為true
,有贊優化了泛化呼叫的效能,為了使用該特性,引入了一個新值result_no_change
- url
Dubbo 服務的地址:dubbo://IP地址:埠
- registryProtocol
Dubbo 註冊中心的協議,設定為ETCD3
- registryAddress
Dubbo 註冊中心的地址
如果是測試 Dubbo 單機水位,則設定 url,註冊中心設定為空;如果是測試 Dubbo 叢集水位,則設定註冊中心(目前支援 ETCD3),url 設定為空。由於目前註冊中心只支援 ETCD3,外掛在 Dubbo 叢集上使用缺乏靈活性,所以我們又實現了客戶端層面的負載均衡,如此便可拋開特定的註冊中心來測試 Dubbo 叢集水位。該特性目前正在內測中。
object DubboProtocol {
val DubboProtocolKey = new ProtocolKey {
type Protocol = DubboProtocol
type Components = DubboComponents
def protocolClass: Class[io.gatling.core.protocol.Protocol] = classOf[DubboProtocol].asInstanceOf[Class[io.gatling.core.protocol.Protocol]]
def defaultProtocolValue(configuration: GatlingConfiguration): DubboProtocol = throw new IllegalStateException("Can't provide a default value for DubboProtocol")
def newComponents(system: ActorSystem, coreComponents: CoreComponents): DubboProtocol => DubboComponents = {
dubboProtocol => DubboComponents(dubboProtocol)
}
}
}
case class DubboProtocol(
protocol: String, //dubbo
generic: String, //泛化呼叫?
url: String, //use url or
registryProtocol: String, //use registry
registryAddress: String //use registry
) extends Protocol {
type Components = DubboComponents
}
複製程式碼
為了方便 Action 中使用上面這些屬性,我們將其裝進了 Gatling 的 ProtocolComponents:
case class DubboComponents(dubboProtocol: DubboProtocol) extends ProtocolComponents {
def onStart: Option[Session => Session] = None
def onExit: Option[Session => Unit] = None
}
複製程式碼
以上就是關於 Protocol 的定義。為了能在 DSL 中配置上述 Protocol,我們定義了 DubboProtocolBuilder,包含了 5 個方法分別設定 Protocol 的 protocol、generic、url、registryProtocol、registryAddress 5 個屬性。
object DubboProtocolBuilderBase {
def protocol(protocol: String) = DubboProtocolBuilderGenericStep(protocol)
}
case class DubboProtocolBuilderGenericStep(protocol: String) {
def generic(generic: String) = DubboProtocolBuilderUrlStep(protocol, generic)
}
case class DubboProtocolBuilderUrlStep(protocol: String, generic: String) {
def url(url: String) = DubboProtocolBuilderRegistryProtocolStep(protocol, generic, url)
}
case class DubboProtocolBuilderRegistryProtocolStep(protocol: String, generic: String, url: String) {
def registryProtocol(registryProtocol: String) = DubboProtocolBuilderRegistryAddressStep(protocol, generic, url, registryProtocol)
}
case class DubboProtocolBuilderRegistryAddressStep(protocol: String, generic: String, url: String, registryProtocol: String) {
def registryAddress(registryAddress: String) = DubboProtocolBuilder(protocol, generic, url, registryProtocol, registryAddress)
}
case class DubboProtocolBuilder(protocol: String, generic: String, url: String, registryProtocol: String, registryAddress: String) {
def build = DubboProtocol(
protocol = protocol,
generic = generic,
url = url,
registryProtocol = registryProtocol,
registryAddress = registryAddress
)
}
複製程式碼
Action
DubboAction 包含了 Duboo 請求邏輯、請求結果校驗邏輯以及壓力控制邏輯,需要擴充套件 ExitableAction 並實現 execute 方法。
DubboAction 類的域 argTypes、argValues 分別是泛化呼叫請求引數型別和請求引數值,需為 Expression[] 型別,這樣當使用資料 Feeder 作為壓測指令碼引數輸入時,可以使用類似 ${args_types}
、${args_values}
這樣的表示式從資料 Feeder 中解析對應欄位的值。
execute 方法必須以非同步方式執行 Dubbo 請求,這樣前一個 Dubbo 請求執行後但還未等響應返回時虛擬使用者就可以通過 AKKA Message 立即發起下一個請求,如此一個虛擬使用者可以在很短的時間內構造大量請求。請求方式方面,相比於泛化呼叫,原生 API 呼叫需要客戶端載入 Dubbo 服務相應的 API 包,但有時候卻拿不到,此外,當被測 Dubbo 應用多了,客戶端需要載入多個 API 包,所以出於使用上的便利性,Dubbo 壓測外掛使用泛化呼叫發起請求。
非同步請求響應後會執行 onComplete 方法,校驗請求結果,並根據校驗結果記錄請求成功或失敗日誌,壓測報告就是使用這些日誌統計計算的。
為了控制壓測時的 RPS,則需要實現 throttle 邏輯。實踐中發現,高併發情況下,泛化呼叫效能遠不如原生 API 呼叫效能,且響應時間成倍增長(如此不能表徵 Dubbo 應用的真正效能),導致 Dubbo 壓測外掛壓力控制不準,解決辦法是優化泛化呼叫效能,使之與原生 API 呼叫的效能相近,請參考**dubbo 泛化呼叫效能優化**。
class DubboAction(
interface: String,
method: String,
argTypes: Expression[Array[String]],
argValues: Expression[Array[Object]],
genericService: GenericService,
checks: List[DubboCheck],
coreComponents: CoreComponents,
throttled: Boolean,
val objectMapper: ObjectMapper,
val next: Action
) extends ExitableAction with NameGen {
override def statsEngine: StatsEngine = coreComponents.statsEngine
override def name: String = genName("dubboRequest")
override def execute(session: Session): Unit = recover(session) {
argTypes(session) flatMap { argTypesArray =>
argValues(session) map { argValuesArray =>
val startTime = System.currentTimeMillis()
val f = Future {
try {
genericService.$invoke(method, argTypes(session).get, argValues(session).get)
} finally {
}
}
f.onComplete {
case Success(result) =>
val endTime = System.currentTimeMillis()
val resultMap = result.asInstanceOf[JMap[String, Any]]
val resultJson = objectMapper.writeValueAsString(resultMap)
val (newSession, error) = Check.check(resultJson, session, checks)
error match {
case None =>
statsEngine.logResponse(session, interface + "." + method, ResponseTimings(startTime, endTime), Status("OK"), None, None)
throttle(newSession(session))
case Some(Failure(errorMessage)) =>
statsEngine.logResponse(session, interface + "." + method, ResponseTimings(startTime, endTime), Status("KO"), None, Some(errorMessage))
throttle(newSession(session).markAsFailed)
}
case FuFailure(e) =>
val endTime = System.currentTimeMillis()
statsEngine.logResponse(session, interface + "." + method, ResponseTimings(startTime, endTime), Status("KO"), None, Some(e.getMessage))
throttle(session.markAsFailed)
}
}
}
}
private def throttle(s: Session): Unit = {
if (throttled) {
coreComponents.throttler.throttle(s.scenario, () => next ! s)
} else {
next ! s
}
}
}
複製程式碼
DubboActionBuilder 則是獲取 Protocol 屬性並初始化 Dubbo 客戶端:
case class DubboActionBuilder(interface: String, method: String, argTypes: Expression[Array[String]], argValues: Expression[Array[Object]], checks: List[DubboCheck]) extends ActionBuilder {
private def components(protocolComponentsRegistry: ProtocolComponentsRegistry): DubboComponents =
protocolComponentsRegistry.components(DubboProtocol.DubboProtocolKey)
override def build(ctx: ScenarioContext, next: Action): Action = {
import ctx._
val protocol = components(protocolComponentsRegistry).dubboProtocol
//Dubbo客戶端配置
val reference = new ReferenceConfig[GenericService]
val application = new ApplicationConfig
application.setName("gatling-dubbo")
reference.setApplication(application)
reference.setProtocol(protocol.protocol)
reference.setGeneric(protocol.generic)
if (protocol.url == "") {
val registry = new RegistryConfig
registry.setProtocol(protocol.registryProtocol)
registry.setAddress(protocol.registryAddress)
reference.setRegistry(registry)
} else {
reference.setUrl(protocol.url)
}
reference.setInterface(interface)
val cache = ReferenceConfigCache.getCache
val genericService = cache.get(reference)
val objectMapper: ObjectMapper = new ObjectMapper()
new DubboAction(interface, method, argTypes, argValues, genericService, checks, coreComponents, throttled, objectMapper, next)
}
}
複製程式碼
LambdaProcessBuilder 則提供了設定 Dubbo 泛化呼叫入參的 DSL 以及接下來要介紹的 Check 部分的 DSL
case class DubboProcessBuilder(interface: String, method: String, argTypes: Expression[Array[String]] = _ => Success(Array.empty[String]), argValues: Expression[Array[Object]] = _ => Success(Array.empty[Object]), checks: List[DubboCheck] = Nil) extends DubboCheckSupport {
def argTypes(argTypes: Expression[Array[String]]): DubboProcessBuilder = copy(argTypes = argTypes)
def argValues(argValues: Expression[Array[Object]]): DubboProcessBuilder = copy(argValues = argValues)
def check(dubboChecks: DubboCheck*): DubboProcessBuilder = copy(checks = checks ::: dubboChecks.toList)
def build(): ActionBuilder = DubboActionBuilder(interface, method, argTypes, argValues, checks)
}
複製程式碼
Check
全鏈路壓測中,我們都使用Json Path
校驗 HTTP 請求結果,Dubbo 壓測外掛中,我們也實現了基於Json Path
的校驗。實現 Check,必須實現 Gatling check 中的 Extender 和 Preparer:
package object dubbo {
type DubboCheck = Check[String]
val DubboStringExtender: Extender[DubboCheck, String] =
(check: DubboCheck) => check
val DubboStringPreparer: Preparer[String, String] =
(result: String) => Success(result)
}
複製程式碼
基於Json Path
的校驗邏輯:
trait DubboJsonPathOfType {
self: DubboJsonPathCheckBuilder[String] =>
def ofType[X: JsonFilter](implicit extractorFactory: JsonPathExtractorFactory) = new DubboJsonPathCheckBuilder[X](path, jsonParsers)
}
object DubboJsonPathCheckBuilder {
val CharsParsingThreshold = 200 * 1000
def preparer(jsonParsers: JsonParsers): Preparer[String, Any] =
response => {
if (response.length() > CharsParsingThreshold || jsonParsers.preferJackson)
jsonParsers.safeParseJackson(response)
else
jsonParsers.safeParseBoon(response)
}
def jsonPath(path: Expression[String])(implicit extractorFactory: JsonPathExtractorFactory, jsonParsers: JsonParsers) =
new DubboJsonPathCheckBuilder[String](path, jsonParsers) with DubboJsonPathOfType
}
class DubboJsonPathCheckBuilder[X: JsonFilter](
private[check] val path: Expression[String],
private[check] val jsonParsers: JsonParsers
)(implicit extractorFactory: JsonPathExtractorFactory)
extends DefaultMultipleFindCheckBuilder[DubboCheck, String, Any, X](
DubboStringExtender,
DubboJsonPathCheckBuilder.preparer(jsonParsers)
) {
import extractorFactory._
def findExtractor(occurrence: Int) = path.map(newSingleExtractor[X](_, occurrence))
def findAllExtractor = path.map(newMultipleExtractor[X])
def countExtractor = path.map(newCountExtractor)
}
複製程式碼
DubboCheckSupport 則提供了設定 jsonPath 表示式的 DSL
trait DubboCheckSupport {
def jsonPath(path: Expression[String])(implicit extractorFactory: JsonPathExtractorFactory, jsonParsers: JsonParsers) =
DubboJsonPathCheckBuilder.jsonPath(path)
}
複製程式碼
- Dubbo 壓測指令碼中可以設定一個或多個 check 校驗請求結果,使用 DSL check 方法*
DSL
trait AwsDsl
提供頂層 DSL。我們還定義了 dubboProtocolBuilder2DubboProtocol、dubboProcessBuilder2ActionBuilder 兩個 Scala 隱式方法,以自動構造 DubboProtocol 和 ActionBuilder。
此外,泛化呼叫中使用的引數型別為 Java 型別,而我們的壓測指令碼使用 Scala 編寫,所以這裡需要做兩種語言間的型別轉換,所以我們定義了 transformJsonDubboData 方法
trait DubboDsl extends DubboCheckSupport {
val Dubbo = DubboProtocolBuilderBase
def dubbo(interface: String, method: String) = DubboProcessBuilder(interface, method)
implicit def dubboProtocolBuilder2DubboProtocol(builder: DubboProtocolBuilder): DubboProtocol = builder.build
implicit def dubboProcessBuilder2ActionBuilder(builder: DubboProcessBuilder): ActionBuilder = builder.build()
def transformJsonDubboData(argTypeName: String, argValueName: String, session: Session): Session = {
session.set(argTypeName, toArray(session(argTypeName).as[JList[String]]))
.set(argValueName, toArray(session(argValueName).as[JList[Any]]))
}
private def toArray[T:ClassTag](value: JList[T]): Array[T] = {
value.asScala.toArray
}
}
複製程式碼
object Predef extends DubboDsl
複製程式碼
Dubbo 壓測指令碼和資料 Feeder 示例
壓測指令碼示例:
import io.gatling.core.Predef._
import io.gatling.dubbo.Predef._
import scala.concurrent.duration._
class DubboTest extends Simulation {
val dubboConfig = Dubbo
.protocol("dubbo")
.generic("true")
//直連某臺Dubbo機器,只單獨壓測一臺機器的水位
.url("dubbo://IP地址:埠")
//或設定註冊中心,壓測該Dubbo應用叢集的水位,支援ETCD3註冊中心
.registryProtocol("")
.registryAddress("")
val jsonFileFeeder = jsonFile("data.json").circular //資料Feeder
val dubboScenario = scenario("load test dubbo")
.forever("repeated") {
feed(jsonFileFeeder)
.exec(session => transformJsonDubboData("args_types1", "args_values1", session))
.exec(dubbo("com.xxx.xxxService", "methodName")
.argTypes("${args_types1}")
.argValues("${args_values1}")
.check(jsonPath("$.code").is("200"))
)
}
setUp(
dubboScenario.inject(atOnceUsers(10))
.throttle(
reachRps(10) in (1 seconds),
holdFor(30 seconds))
).protocols(dubboConfig)
}
複製程式碼
data.json 示例:
[
{
"args_types1": ["com.xxx.xxxDTO"],
"args_values1": [{
"field1": "111",
"field2": "222",
"field3": "333"
}]
}
]
複製程式碼
Dubbo 壓測報告示例
我的系列部落格
混沌工程 - 軟體系統高可用、彈性化的必由之路
非同步系統的兩種測試方法
我的其他測試相關開源專案
捉蟲記:方便產品、開發、測試三方協同自測的管理工具
招聘
有贊測試組在持續招人中,大量崗位空缺,只要你來,就能幫你點亮全棧開發技能樹,有意向換工作的同學可以發簡歷到 sunjun【@】youzan.com