volodymyr-babak
10 years ago
10 changed files with 0 additions and 657 deletions
@ -1,99 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.paho.client.mqttv3.*; |
|||
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class MqttStressTestClient { |
|||
|
|||
@Getter |
|||
private final String deviceToken; |
|||
@Getter |
|||
private final String clientId; |
|||
private final MqttClientPersistence persistence; |
|||
private final MqttAsyncClient client; |
|||
private final ResultAccumulator results; |
|||
|
|||
public MqttStressTestClient(ResultAccumulator results, String brokerUri, String deviceToken) throws MqttException { |
|||
this.results = results; |
|||
this.clientId = MqttAsyncClient.generateClientId(); |
|||
this.deviceToken = deviceToken; |
|||
this.persistence = new MemoryPersistence(); |
|||
this.client = new MqttAsyncClient(brokerUri, clientId, persistence); |
|||
} |
|||
|
|||
public IMqttToken connect() throws MqttException { |
|||
MqttConnectOptions options = new MqttConnectOptions(); |
|||
options.setUserName(deviceToken); |
|||
return client.connect(options, null, new IMqttActionListener() { |
|||
@Override |
|||
public void onSuccess(IMqttToken iMqttToken) { |
|||
log.info("OnSuccess"); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(IMqttToken iMqttToken, Throwable e) { |
|||
log.info("OnFailure", e); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public void disconnect() throws MqttException { |
|||
client.disconnect(); |
|||
} |
|||
|
|||
|
|||
|
|||
public void warmUp(byte[] data) throws MqttException { |
|||
MqttMessage msg = new MqttMessage(data); |
|||
client.publish("v1/devices/me/telemetry", msg, null, new IMqttActionListener() { |
|||
@Override |
|||
public void onSuccess(IMqttToken asyncActionToken) { |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(IMqttToken asyncActionToken, Throwable exception) { |
|||
} |
|||
}).waitForCompletion(); |
|||
} |
|||
|
|||
|
|||
public void publishTelemetry(byte[] data) throws MqttException { |
|||
long sendTime = System.currentTimeMillis(); |
|||
MqttMessage msg = new MqttMessage(data); |
|||
client.publish("v1/devices/me/telemetry", msg, null, new IMqttActionListener() { |
|||
@Override |
|||
public void onSuccess(IMqttToken asyncActionToken) { |
|||
long ackTime = System.currentTimeMillis(); |
|||
results.onResult(true, ackTime - sendTime); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(IMqttToken asyncActionToken, Throwable exception) { |
|||
long failTime = System.currentTimeMillis(); |
|||
results.onResult(false, failTime - sendTime); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
@ -1,135 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.paho.client.mqttv3.IMqttToken; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
|
|||
import java.nio.charset.StandardCharsets; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.Callable; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.ScheduledFuture; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class MqttStressTestTool { |
|||
|
|||
private static byte[] data = "{\"longKey\":73}".getBytes(StandardCharsets.UTF_8); |
|||
private static ResultAccumulator results = new ResultAccumulator(); |
|||
private static List<MqttStressTestClient> clients = new ArrayList<>(); |
|||
private static List<IMqttToken> connectTokens = new ArrayList<>(); |
|||
|
|||
public static void main(String[] args) throws Exception { |
|||
TestParams params = new TestParams(); |
|||
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10); |
|||
|
|||
if (params.getDuration() % params.getIterationInterval() != 0) { |
|||
throw new IllegalArgumentException("Test Duration % Iteration Interval != 0"); |
|||
} |
|||
|
|||
if ((params.getIterationInterval() * 1000) % params.getDeviceCount() != 0) { |
|||
throw new IllegalArgumentException("Iteration Interval % Device Count != 0"); |
|||
} |
|||
|
|||
createDevices(params); |
|||
|
|||
long startTime = System.currentTimeMillis(); |
|||
int iterationsCount = (int) (params.getDuration() / params.getIterationInterval()); |
|||
int subIterationMicroSeconds = (int) ((params.getIterationInterval() * 1000) / params.getDeviceCount()); |
|||
|
|||
List<ScheduledFuture<Void>> iterationFutures = new ArrayList<>(); |
|||
for (int i = 0; i < iterationsCount; i++) { |
|||
long delay = i * params.getIterationInterval(); |
|||
iterationFutures.add(scheduler.schedule((Callable<Void>) () -> { |
|||
long sleepMicroSeconds = 0L; |
|||
for (MqttStressTestClient client : clients) { |
|||
client.publishTelemetry(data); |
|||
sleepMicroSeconds += subIterationMicroSeconds; |
|||
if (sleepMicroSeconds > 1000) { |
|||
Thread.sleep(sleepMicroSeconds / 1000); |
|||
sleepMicroSeconds = sleepMicroSeconds % 1000; |
|||
} |
|||
} |
|||
return null; |
|||
}, delay, TimeUnit.MILLISECONDS)); |
|||
} |
|||
|
|||
for (ScheduledFuture<Void> future : iterationFutures) { |
|||
future.get(); |
|||
} |
|||
|
|||
Thread.sleep(1000); |
|||
|
|||
for (MqttStressTestClient client : clients) { |
|||
client.disconnect(); |
|||
} |
|||
log.info("Results: {} took {}ms", results, System.currentTimeMillis() - startTime); |
|||
scheduler.shutdownNow(); |
|||
} |
|||
|
|||
/** |
|||
* Returns list of device credential IDs |
|||
* |
|||
* @param params |
|||
* @return |
|||
* @throws Exception |
|||
*/ |
|||
public static List<String> createDevices(TestParams params) throws Exception { |
|||
AtomicLong value = new AtomicLong(Long.MAX_VALUE); |
|||
log.info("value: {} ", value.incrementAndGet()); |
|||
|
|||
RestClient restClient = new RestClient(params.getRestApiUrl()); |
|||
restClient.login(params.getUsername(), params.getPassword()); |
|||
|
|||
List<String> deviceCredentialsIds = new ArrayList<>(); |
|||
for (int i = 0; i < params.getDeviceCount(); i++) { |
|||
Device device = restClient.createDevice("Device " + UUID.randomUUID()); |
|||
DeviceCredentials credentials = restClient.getCredentials(device.getId()); |
|||
String[] mqttUrls = params.getMqttUrls(); |
|||
String mqttURL = mqttUrls[i % mqttUrls.length]; |
|||
MqttStressTestClient client = new MqttStressTestClient(results, mqttURL, credentials.getCredentialsId()); |
|||
|
|||
deviceCredentialsIds.add(credentials.getCredentialsId()); |
|||
|
|||
connectTokens.add(client.connect()); |
|||
clients.add(client); |
|||
} |
|||
|
|||
for (IMqttToken tokens : connectTokens) { |
|||
tokens.waitForCompletion(); |
|||
} |
|||
|
|||
for (MqttStressTestClient client : clients) { |
|||
client.warmUp(data); |
|||
} |
|||
|
|||
Thread.sleep(1000); |
|||
|
|||
return deviceCredentialsIds; |
|||
} |
|||
} |
|||
@ -1,77 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.http.HttpRequest; |
|||
import org.springframework.http.ResponseEntity; |
|||
import org.springframework.http.client.ClientHttpRequestExecution; |
|||
import org.springframework.http.client.ClientHttpRequestInterceptor; |
|||
import org.springframework.http.client.ClientHttpResponse; |
|||
import org.springframework.http.client.support.HttpRequestWrapper; |
|||
import org.springframework.web.client.RestTemplate; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
|
|||
import java.io.IOException; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@RequiredArgsConstructor |
|||
public class RestClient implements ClientHttpRequestInterceptor { |
|||
private static final String JWT_TOKEN_HEADER_PARAM = "X-Authorization"; |
|||
private final RestTemplate restTemplate = new RestTemplate(); |
|||
private String token; |
|||
private final String baseURL; |
|||
|
|||
public void login(String username, String password) { |
|||
Map<String, String> loginRequest = new HashMap<>(); |
|||
loginRequest.put("username", username); |
|||
loginRequest.put("password", password); |
|||
ResponseEntity<JsonNode> tokenInfo = restTemplate.postForEntity(baseURL + "/api/auth/login", loginRequest, JsonNode.class); |
|||
this.token = tokenInfo.getBody().get("token").asText(); |
|||
restTemplate.setInterceptors(Collections.singletonList(this)); |
|||
} |
|||
|
|||
public Device createDevice(String name) { |
|||
Device device = new Device(); |
|||
device.setName(name); |
|||
return restTemplate.postForEntity(baseURL + "/api/device", device, Device.class).getBody(); |
|||
} |
|||
|
|||
public DeviceCredentials getCredentials(DeviceId id) { |
|||
return restTemplate.getForEntity(baseURL + "/api/device/" + id.getId().toString() + "/credentials", DeviceCredentials.class).getBody(); |
|||
} |
|||
|
|||
public RestTemplate getRestTemplate() { |
|||
return restTemplate; |
|||
} |
|||
|
|||
@Override |
|||
public ClientHttpResponse intercept(HttpRequest request, byte[] bytes, ClientHttpRequestExecution execution) throws IOException { |
|||
HttpRequest wrapper = new HttpRequestWrapper(request); |
|||
wrapper.getHeaders().set(JWT_TOKEN_HEADER_PARAM, "Bearer " + token); |
|||
return execution.execute(wrapper, bytes); |
|||
} |
|||
|
|||
} |
|||
@ -1,88 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class ResultAccumulator { |
|||
|
|||
private AtomicLong minTime = new AtomicLong(Long.MAX_VALUE); |
|||
private AtomicLong maxTime = new AtomicLong(Long.MIN_VALUE); |
|||
private AtomicLong timeSpentCount = new AtomicLong(); |
|||
private AtomicInteger successCount = new AtomicInteger(); |
|||
private AtomicInteger errorCount = new AtomicInteger(); |
|||
|
|||
public void onResult(boolean success, long timeSpent) { |
|||
if (success) { |
|||
successCount.incrementAndGet(); |
|||
} else { |
|||
errorCount.incrementAndGet(); |
|||
} |
|||
timeSpentCount.addAndGet(timeSpent); |
|||
|
|||
while (!setMax(timeSpent)) ; |
|||
while (!setMin(timeSpent)) ; |
|||
} |
|||
|
|||
private boolean setMax(long timeSpent) { |
|||
long curMax = maxTime.get(); |
|||
long newMax = Math.max(curMax, timeSpent); |
|||
return maxTime.compareAndSet(curMax, newMax); |
|||
} |
|||
|
|||
private boolean setMin(long timeSpent) { |
|||
long curMin = minTime.get(); |
|||
long newMin = Math.min(curMin, timeSpent); |
|||
return minTime.compareAndSet(curMin, newMin); |
|||
} |
|||
|
|||
|
|||
public int getSuccessCount() { |
|||
return successCount.get(); |
|||
} |
|||
|
|||
public int getErrorCount() { |
|||
return errorCount.get(); |
|||
} |
|||
|
|||
public long getTimeSpent() { |
|||
return timeSpentCount.get(); |
|||
} |
|||
|
|||
public double getAvgTimeSpent() { |
|||
return ((double) getTimeSpent()) / (getSuccessCount() + getErrorCount()); |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "Result {" + |
|||
"successCount=" + getSuccessCount() + |
|||
", errorCount=" + getErrorCount() + |
|||
", totalTime=" + getTimeSpent() + |
|||
", avgTime=" + getAvgTimeSpent() + |
|||
", minTime=" + minTime.get() + |
|||
", maxTime=" + maxTime.get() + |
|||
'}'; |
|||
} |
|||
} |
|||
@ -1,74 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.io.FileInputStream; |
|||
import java.io.IOException; |
|||
import java.net.URL; |
|||
import java.util.Properties; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Slf4j |
|||
public class TestParams { |
|||
static final String TEST_PROPERTIES = "test.properties"; |
|||
static final long DEFAULT_TEST_DURATION = TimeUnit.SECONDS.toMillis(1); |
|||
static final long DEFAULT_TEST_INTERVAL = TimeUnit.MILLISECONDS.toMillis(100); |
|||
static final int DEFAULT_DEVICE_COUNT = 2000; |
|||
static final String DEFAULT_REST_URL = "http://localhost:8080"; |
|||
static final String DEFAULT_MQTT_URLS = "tcp://localhost:1883"; |
|||
static final String DEFAULT_USERNAME = "tenant@thingsboard.org"; |
|||
static final String DEFAULT_PASSWORD = "tenant"; |
|||
|
|||
private Properties params = new Properties(); |
|||
|
|||
public TestParams() throws IOException { |
|||
try { |
|||
params.load(TestParams.class.getClassLoader().getResourceAsStream(TEST_PROPERTIES)); |
|||
} catch (Exception e) { |
|||
log.warn("Failed to read " + TEST_PROPERTIES); |
|||
} |
|||
} |
|||
|
|||
public long getDuration() { |
|||
return Long.valueOf(params.getProperty("durationMs", Long.toString(DEFAULT_TEST_DURATION))); |
|||
} |
|||
|
|||
public long getIterationInterval() { |
|||
return Long.valueOf(params.getProperty("iterationIntervalMs", Long.toString(DEFAULT_TEST_INTERVAL))); |
|||
} |
|||
|
|||
public int getDeviceCount() { |
|||
return Integer.valueOf(params.getProperty("deviceCount", Integer.toString(DEFAULT_DEVICE_COUNT))); |
|||
} |
|||
|
|||
public String getRestApiUrl() { |
|||
return params.getProperty("restUrl", DEFAULT_REST_URL); |
|||
} |
|||
|
|||
public String[] getMqttUrls() { |
|||
return params.getProperty("mqttUrls", DEFAULT_MQTT_URLS).split(","); |
|||
} |
|||
|
|||
public String getUsername() { |
|||
return params.getProperty("username", DEFAULT_USERNAME); |
|||
} |
|||
|
|||
public String getPassword() { |
|||
return params.getProperty("password", DEFAULT_PASSWORD); |
|||
} |
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
Copyright © 2016-2017 The Thingsboard Authors |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration> |
|||
|
|||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="io.gatling.core.action.Pause" level="WARN"/> |
|||
<logger name="org.thingsboard" level="INFO" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="STDOUT"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -1,5 +0,0 @@ |
|||
restUrl=http://localhost:8080 |
|||
mqttUrls=tcp://localhost:1883 |
|||
deviceCount=100 |
|||
durationMs=60000 |
|||
iterationIntervalMs=1000 |
|||
@ -1,55 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.client.tools |
|||
|
|||
import com.github.mnogu.gatling.mqtt.Predef._ |
|||
import io.gatling.core.Predef._ |
|||
import org.fusesource.mqtt.client.QoS |
|||
import scala.collection.JavaConverters._ |
|||
|
|||
import scala.concurrent.duration._ |
|||
|
|||
class MqttSimulation extends Simulation { |
|||
|
|||
val testParams = new TestParams() |
|||
|
|||
val deviceCredentialsIds: Array[String] = MqttStressTestTool.createDevices(testParams).asScala.toArray |
|||
|
|||
val mqttConf = mqtt |
|||
.host("tcp://localhost:1883") |
|||
.userName("${deviceCredentialsId}") |
|||
|
|||
val connect = exec(mqtt("connect") |
|||
.connect()) |
|||
|
|||
val publish = repeat(100) { |
|||
exec(mqtt("publish") |
|||
.publish("v1/devices/me/telemetry", "{\"key1\":\"value1\", \"key2\":\"value2\"}", QoS.AT_LEAST_ONCE, retain = false)) |
|||
.pause(100 milliseconds) |
|||
} |
|||
|
|||
val deviceCredentialsIdsFeeder = deviceCredentialsIds.map( x => {Map("deviceCredentialsId" -> x)}) |
|||
|
|||
val scn = scenario("Scenario Name") |
|||
.feed(deviceCredentialsIdsFeeder) |
|||
.exec(connect, publish) |
|||
|
|||
setUp( |
|||
scn |
|||
.inject(constantUsersPerSec(testParams.getDeviceCount) during (1 seconds)) |
|||
).protocols(mqttConf) |
|||
|
|||
} |
|||
Loading…
Reference in new issue