RabbitMQ collector

.travis.yml

@@ -19,6 +19,7 @@ jdk:
 
 services:
   - docker
+  - rabbitmq
 
 before_install:
   # parameters used during a release

pom.xml

@@ -73,7 +73,7 @@
     <assertj.version>3.8.0</assertj.version>
     <hamcrest.version>1.3</hamcrest.version>
     <okhttp.version>3.8.1</okhttp.version>
-    <testcontainers.version>1.4.1</testcontainers.version>
+    <testcontainers.version>1.4.2</testcontainers.version>
 
     <animal-sniffer-maven-plugin.version>1.15</animal-sniffer-maven-plugin.version>
     <maven-compiler-plugin.version>3.6.1</maven-compiler-plugin.version>
@@ -282,6 +282,18 @@
         <version>${project.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>${project.groupId}</groupId>
+        <artifactId>zipkin-collector-rabbitmq</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>${project.groupId}</groupId>
+        <artifactId>zipkin-autoconfigure-collector-rabbitmq</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>${project.groupId}</groupId>
         <artifactId>zipkin-collector-scribe</artifactId>
@@ -318,6 +330,13 @@
       </dependency>
       <!-- End spring-boot-dependencies overrides -->
 
+      <!-- testcontainers overrides jnr-posix which conflicts w/ cassandra -->
+      <dependency>
+        <groupId>com.github.jnr</groupId>
+        <artifactId>jnr-posix</artifactId>
+        <version>3.0.41</version>
+      </dependency>
+
       <dependency>
         <groupId>com.squareup.okhttp3</groupId>
         <artifactId>okhttp</artifactId>

zipkin-autoconfigure/collector-rabbitmq/pom.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright 2015-2017 The OpenZipkin 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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>
+
+  <parent>
+    <groupId>io.zipkin.java</groupId>
+    <artifactId>zipkin-autoconfigure</artifactId>
+    <version>2.0.2-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>zipkin-autoconfigure-collector-rabbitmq</artifactId>
+  <name>Auto Configuration: RabbitMQ Collector</name>
+
+  <properties>
+    <main.basedir>${project.basedir}/../..</main.basedir>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-collector-rabbitmq</artifactId>
+    </dependency>
+  </dependencies>
+
+</project>

zipkin-autoconfigure/collector-rabbitmq/src/main/java/zipkin/autoconfigure/collector/rabbitmq/ZipkinRabbitMQCollectorAutoConfiguration.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright 2015-2017 The OpenZipkin 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 zipkin.autoconfigure.collector.rabbitmq;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import zipkin.collector.CollectorMetrics;
+import zipkin.collector.CollectorSampler;
+import zipkin.collector.rabbitmq.RabbitMQCollector;
+import zipkin.storage.StorageComponent;
+
+/**
+ * Auto-configuration for {@link RabbitMQCollector}.
+ */
+@Configuration
+@Conditional(ZipkinRabbitMQCollectorAutoConfiguration.RabbitMqAddressesSet.class)
+@EnableConfigurationProperties(ZipkinRabbitMQCollectorProperties.class)
+public class ZipkinRabbitMQCollectorAutoConfiguration {
+
+  @Bean(initMethod = "start") RabbitMQCollector rabbitMq(
+    ZipkinRabbitMQCollectorProperties properties,
+    CollectorSampler sampler, CollectorMetrics metrics, StorageComponent storage) {
+    return properties.toBuilder().sampler(sampler).metrics(metrics).storage(storage).build();
+  }
+
+  /**
+   * This condition passes when {@link ZipkinRabbitMQCollectorProperties#getAddresses()} is set to
+   * non-empty.
+   *
+   * <p>This is here because the yaml defaults this property to empty like this, and Spring Boot
+   * doesn't have an option to treat empty properties as unset.
+   *
+   * <pre>{@code
+   * addresses: ${RABBIT_ADDRESSES:}
+   * }</pre>
+   */
+  static final class RabbitMqAddressesSet implements Condition {
+    @Override public boolean matches(ConditionContext context, AnnotatedTypeMetadata a) {
+      return !isEmpty(context.getEnvironment().getProperty("zipkin.collector.rabbitmq.addresses"));
+    }
+
+    private static boolean isEmpty(String s) {
+      return s == null || s.isEmpty();
+    }
+  }
+}

zipkin-autoconfigure/collector-rabbitmq/src/main/java/zipkin/autoconfigure/collector/rabbitmq/ZipkinRabbitMQCollectorProperties.java

@@ -0,0 +1,112 @@
+/**
+ * Copyright 2015-2017 The OpenZipkin 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 zipkin.autoconfigure.collector.rabbitmq;
+
+import com.rabbitmq.client.ConnectionFactory;
+import java.util.List;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import zipkin.collector.rabbitmq.RabbitMQCollector;
+
+/**
+ * Properties for configuring and building a {@link RabbitMQCollector}.
+ */
+@ConfigurationProperties("zipkin.collector.rabbitmq")
+public class ZipkinRabbitMQCollectorProperties {
+
+  /** RabbitMQ server addresses in the form of a (comma-separated) list of host:port pairs */
+  private List<String> addresses;
+  /** Number of concurrent consumers */
+  private Integer concurrency = 1;
+  /** TCP connection timeout in milliseconds */
+  private Integer connectionTimeout;
+  /** RabbitMQ user password */
+  private String password;
+  /** RabbitMQ queue from which to collect the Zipkin spans */
+  private String queue;
+  /** RabbitMQ username */
+  private String username;
+  /** RabbitMQ virtual host */
+  private String virtualHost;
+
+  public List<String> getAddresses() {
+    return addresses;
+  }
+
+  public void setAddresses(List<String> addresses) {
+    this.addresses = addresses;
+  }
+
+  public int getConcurrency() {
+    return concurrency;
+  }
+
+  public void setConcurrency(int concurrency) {
+    this.concurrency = concurrency;
+  }
+
+  public Integer getConnectionTimeout() {
+    return connectionTimeout;
+  }
+
+  public void setConnectionTimeout(Integer connectionTimeout) {
+    this.connectionTimeout = connectionTimeout;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+
+  public void setPassword(String password) {
+    this.password = password;
+  }
+
+  public String getQueue() {
+    return queue;
+  }
+
+  public void setQueue(String queue) {
+    this.queue = queue;
+  }
+
+  public String getUsername() {
+    return username;
+  }
+
+  public void setUsername(String username) {
+    this.username = username;
+  }
+
+  public String getVirtualHost() {
+    return virtualHost;
+  }
+
+  public void setVirtualHost(String virtualHost) {
+    this.virtualHost = virtualHost;
+  }
+
+  public RabbitMQCollector.Builder toBuilder() {
+    final RabbitMQCollector.Builder result = RabbitMQCollector.builder();
+    ConnectionFactory connectionFactory = new ConnectionFactory();
+    if (addresses != null) result.addresses(addresses);
+    if (concurrency != null) result.concurrency(concurrency);
+    if (connectionTimeout != null) connectionFactory.setConnectionTimeout(connectionTimeout);
+    if (password != null) connectionFactory.setPassword(password);
+    if (queue != null) result.queue(queue);
+    if (username != null) connectionFactory.setUsername(username);
+    if (virtualHost != null) connectionFactory.setVirtualHost(virtualHost);
+    result.connectionFactory(connectionFactory);
+    return result;
+  }
+
+}

zipkin-autoconfigure/collector-rabbitmq/src/main/java/zipkin/autoconfigure/collector/rabbitmq/package-info.java

@@ -0,0 +1,15 @@
+/**
+ * Copyright 2015-2017 The OpenZipkin 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.
+ */
+@javax.annotation.ParametersAreNonnullByDefault
+package zipkin.autoconfigure.collector.rabbitmq;

zipkin-autoconfigure/collector-rabbitmq/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+  zipkin.autoconfigure.collector.rabbitmq.ZipkinRabbitMQCollectorAutoConfiguration

zipkin-autoconfigure/collector-rabbitmq/src/test/java/zipkin/collector/rabbitmq/ZipkinRabbitMQCollectorAutoConfigurationTest.java

@@ -0,0 +1,98 @@
+/**
+ * Copyright 2015-2017 The OpenZipkin 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 zipkin.collector.rabbitmq;
+
+import org.junit.After;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import zipkin.autoconfigure.collector.rabbitmq.ZipkinRabbitMQCollectorAutoConfiguration;
+import zipkin.collector.CollectorMetrics;
+import zipkin.collector.CollectorSampler;
+import zipkin.storage.InMemoryStorage;
+import zipkin.storage.StorageComponent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.util.EnvironmentTestUtils.addEnvironment;
+
+public class ZipkinRabbitMQCollectorAutoConfigurationTest {
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  private AnnotationConfigApplicationContext context;
+
+  @After
+  public void close() {
+    if (context != null) {
+      context.close();
+    }
+  }
+
+  @Test
+  public void doesNotProvideCollectorComponent_whenAddressNotSet() {
+    context = new AnnotationConfigApplicationContext();
+    context.register(PropertyPlaceholderAutoConfiguration.class,
+        ZipkinRabbitMQCollectorAutoConfiguration.class, InMemoryConfiguration.class);
+    context.refresh();
+
+    thrown.expect(NoSuchBeanDefinitionException.class);
+    context.getBean(RabbitMQCollector.class);
+  }
+
+  @Test
+  public void doesNotProvideCollectorComponent_whenAddressesIsEmptyString() {
+    context = new AnnotationConfigApplicationContext();
+    addEnvironment(context, "zipkin.collector.rabbitmq.addresses:");
+    context.register(PropertyPlaceholderAutoConfiguration.class,
+        ZipkinRabbitMQCollectorAutoConfiguration.class, InMemoryConfiguration.class);
+    context.refresh();
+
+    thrown.expect(NoSuchBeanDefinitionException.class);
+    context.getBean(RabbitMQCollector.class);
+  }
+
+  @Test
+  @Ignore
+  public void providesCollectorComponent_whenAddressesSet() {
+    context = new AnnotationConfigApplicationContext();
+    addEnvironment(context, "zipkin.collector.rabbitmq.addresses=localhost:5672");
+    context.register(PropertyPlaceholderAutoConfiguration.class,
+        ZipkinRabbitMQCollectorAutoConfiguration.class, InMemoryConfiguration.class);
+    context.refresh();
+
+    assertThat(context.getBean(RabbitMQCollector.class)).isNotNull();
+  }
+
+  @Configuration
+  static class InMemoryConfiguration {
+    @Bean CollectorSampler sampler() {
+      return CollectorSampler.ALWAYS_SAMPLE;
+    }
+
+    @Bean CollectorMetrics metrics() {
+      return CollectorMetrics.NOOP_METRICS;
+    }
+
+    @Bean StorageComponent storage() {
+      return new InMemoryStorage();
+    }
+  }
+}