SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrate throughput sink

Browse Source 
Resolves https://github.com/spring-cloud/stream-applications/issues/25
This commit is contained in:
Soby Chacko
2020-05-12 13:28:15 -04:00
committed by Artem Bilan
parent 2b1089a009
commit a7b5ee4ce9
7 changed files with 336 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/sink/pom.xml
View File

@@ -21,5 +21,6 @@
<module>rabbit-sink</module>
<module>router-sink</module>
<module>sftp-sink</module>
<module>throughput-sink</module>
</modules>
</project>

14
applications/sink/throughput-sink/README.adoc Normal file
View File

@@ -0,0 +1,14 @@
//tag::ref-doc[]
= Throughput Sink
Sink that will count messages and log the observed throughput at a selected interval.
== Options
The **$$throughput$$** $$sink$$ has the following options:
//tag::configuration-properties[]
$$throughput.report-every-ms$$:: $$how often to report.$$ *($$Integer$$, default: `$$1000$$`)*
//end::configuration-properties[]
//end::ref-doc[]

91
applications/sink/throughput-sink/pom.xml Normal file
View File

@@ -0,0 +1,91 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>throughput-sink</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>throughput-sink</name>
<description>throughput sink apps</description>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>throughput</name>
<type>sink</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.stream.app.sink.throughput.ThroughputConsumerConfiguration.class</configClass>
<functionDefinition>throughputConsumer</functionDefinition>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>throughput-sink</artifactId>
<version>${java-functions.version}</version>
</dependency>
</dependencies>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</repositories>
</project>

129
applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkConfiguration.java Normal file
View File

@@ -0,0 +1,129 @@
/*
* Copyright 2015-2020 the original author or 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
*
* https://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.springframework.cloud.stream.app.sink.throughput;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
@Configuration
@EnableConfigurationProperties({ThroughputSinkProperties.class})
public class ThroughputSinkConfiguration {
private final Log logger = LogFactory.getLog(getClass());
private final AtomicLong counter = new AtomicLong();
private final AtomicLong start = new AtomicLong(-1);
private final AtomicLong bytes = new AtomicLong(-1);
private final AtomicLong intermediateCounter = new AtomicLong();
private final AtomicLong intermediateBytes = new AtomicLong();
private final TimeUnit timeUnit = TimeUnit.SECONDS;
private final ExecutorService executorService = Executors.newFixedThreadPool(1);
private volatile boolean reportBytes = false;
@Autowired
private volatile ThroughputSinkProperties properties;
@Bean
public Consumer<Message<?>> throughputConsumer() {
return message -> {
if (start.get() == -1L) {
synchronized (start) {
if (start.get() == -1L) {
// assume a homogeneous message structure - this is intended for
// performance tests so we can assume that the messages are similar;
// therefore we'll do our reporting based on the first message
Object payload = message.getPayload();
if (payload instanceof byte[] || payload instanceof String) {
reportBytes = true;
}
start.set(System.currentTimeMillis());
executorService.execute(new ReportStats());
}
}
}
intermediateCounter.incrementAndGet();
if (reportBytes) {
Object payload = message.getPayload();
if (payload instanceof byte[]) {
intermediateBytes.addAndGet(((byte[]) payload).length);
}
else if (payload instanceof String) {
intermediateBytes.addAndGet((((String) payload).getBytes()).length);
}
}
};
}
private class ReportStats implements Runnable {
@Override
public void run() {
int reportEveryMs = properties.getReportEveryMs();
long intervalStart = System.currentTimeMillis();
try {
Thread.sleep(reportEveryMs);
long timeNow = System.currentTimeMillis();
long currentCounter = intermediateCounter.getAndSet(0L);
long currentBytes = intermediateBytes.getAndSet(0L);
long totalCounter = counter.addAndGet(currentCounter);
long totalBytes = bytes.addAndGet(currentBytes);
logger.info(
String.format("Messages: %10d in %5.2f%s = %11.2f/s",
currentCounter,
(timeNow - intervalStart) / 1000.0, timeUnit, ((double) currentCounter * 1000 / reportEveryMs)));
logger.info(
String.format("Messages: %10d in %5.2f%s = %11.2f/s",
totalCounter, (timeNow - start.get()) / 1000.0, timeUnit,
((double) totalCounter * 1000 / (timeNow - start.get()))));
if (reportBytes) {
logger.info(
String.format("Throughput: %12d in %5.2f%s = %11.2fMB/s, ",
currentBytes,
(timeNow - intervalStart) / 1000.0, timeUnit,
((currentBytes / (1024.0 * 1024)) * 1000 / reportEveryMs)));
logger.info(
String.format("Throughput: %12d in %5.2f%s = %11.2fMB/s",
totalBytes, (timeNow - start.get()) / 1000.0, timeUnit,
((totalBytes / (1024.0 * 1024)) * 1000 / (timeNow - start.get()))));
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.warn("Thread interrupted", e);
}
}
}
}

42
applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkProperties.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2013-2020 the original author or 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
*
* https://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.springframework.cloud.stream.app.sink.throughput;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Holds configuration options for the throughput Sink.
*
* @author Glenn Renfro
*/
@ConfigurationProperties("throughput")
public class ThroughputSinkProperties {
/**
* how often to report.
*/
private int reportEveryMs = 1000;
public int getReportEveryMs() {
return reportEveryMs;
}
public void setReportEveryMs(int reportEveryMs) {
this.reportEveryMs = reportEveryMs;
}
}

2
applications/sink/throughput-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -0,0 +1,2 @@
configuration-properties.classes=org.springframework.cloud.stream.app.sink.throughput.ThroughputSinkProperties

57
applications/sink/throughput-sink/src/test/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkTests.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2015-2020 the original author or 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
*
* https://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.springframework.cloud.stream.app.sink.throughput;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@ExtendWith(OutputCaptureExtension.class)
public class ThroughputSinkTests {
@Test
public void testThroughputSink(CapturedOutput output) throws Exception {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration
.getCompleteConfiguration(ThroughputSinkTestConfiguration.class))
.web(WebApplicationType.NONE)
.run()) {
final Message<String> message = MessageBuilder.withPayload("hello").build();
InputDestination source = context.getBean(InputDestination.class);
source.send(message);
Awaitility.await().until(output::getOut, value -> value.contains("Messages:") && value.contains("Throughput:"));
}
}
@EnableAutoConfiguration
@Import(ThroughputSinkConfiguration.class)
public static class ThroughputSinkTestConfiguration {
}
}