diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index df07b03e..203b3285 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -21,5 +21,6 @@ rabbit-sink router-sink sftp-sink + throughput-sink diff --git a/applications/sink/throughput-sink/README.adoc b/applications/sink/throughput-sink/README.adoc new file mode 100644 index 00000000..1a077a49 --- /dev/null +++ b/applications/sink/throughput-sink/README.adoc @@ -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[] diff --git a/applications/sink/throughput-sink/pom.xml b/applications/sink/throughput-sink/pom.xml new file mode 100644 index 00000000..139c3541 --- /dev/null +++ b/applications/sink/throughput-sink/pom.xml @@ -0,0 +1,91 @@ + + + 4.0.0 + throughput-sink + 3.0.0-SNAPSHOT + throughput-sink + throughput sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.awaitility + awaitility + test + + + junit + junit + + + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + throughput + sink + ${project.version} + org.springframework.cloud.stream.app.sink.throughput.ThroughputConsumerConfiguration.class + throughputConsumer + + + + org.springframework.cloud.stream.app + throughput-sink + ${java-functions.version} + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkConfiguration.java b/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkConfiguration.java new file mode 100644 index 00000000..e70552a4 --- /dev/null +++ b/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkConfiguration.java @@ -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> 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); + } + } + } +} diff --git a/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkProperties.java b/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkProperties.java new file mode 100644 index 00000000..5fccf2d4 --- /dev/null +++ b/applications/sink/throughput-sink/src/main/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkProperties.java @@ -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; + } + +} diff --git a/applications/sink/throughput-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/throughput-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..6728587f --- /dev/null +++ b/applications/sink/throughput-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.stream.app.sink.throughput.ThroughputSinkProperties + diff --git a/applications/sink/throughput-sink/src/test/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkTests.java b/applications/sink/throughput-sink/src/test/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkTests.java new file mode 100644 index 00000000..53c7b1b0 --- /dev/null +++ b/applications/sink/throughput-sink/src/test/java/org/springframework/cloud/stream/app/sink/throughput/ThroughputSinkTests.java @@ -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 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 { + } +}